使用 Azure 服务总线主题时从 _error 队列中移动消息答案

【问题标题】：Moving messages from _error queue when using Azure Service Bus Topics使用 Azure 服务总线主题时从 _error 队列中移动消息
【发布时间】：2020-07-08 23:31:20
【问题描述】：

我正在尝试使用 Azure 服务总线主题来允许服务使用 MassTransit 使用消息。一条消息发布到一个主题，每个订阅该主题的服务都会收到该消息的副本。每种消息类型都有一个主题。

问题是一旦解决了异常的来源，如何处理错误。一旦消息出现故障并最终进入 _error 队列，我想在消息和/或服务修复后将消息移回以进行处理。我无法将消息从 _error 队列移动到主题，因为该主题上的每个服务都会再次收到消息。

我尝试使用名为 _errorrecovery 的 ReceiveEndpoint 方法创建第二个队列，但这样做会导致队列订阅主题，这意味着 _errorrecovery 队列会获取发布到该主题的每条消息。

我想知道是否有一种方法可以使用 MassTransit 设置一个队列，该队列将只处理该队列中的消息而不添加额外的订阅。

这是我当前构建主题的设置。

TEvent 是消息类型，TConsumer 是该消息类型的关联 IConsumer 实现。

  public void ConfigureType<TEvent, TConsumer>(IServiceBusBusFactoryConfigurator busConfig, Container container, MessageHandlingOptions options) where TConsumer : class, IConsumer
        {
            string subName = NameHelper.GetSubscriptionName(@namespace, _serviceName);
            var topicName = NameHelper.GetTopicName(@namespace, typeof(TEvent));

            busConfig.SubscriptionEndpoint(subName, topicName, configurator =>
            {
                configurator.ConfigureConsumer(container, typeof(TConsumer));
                if (!(options is null))
                {
                    ConfigureRetry(configurator, options);
                }
            });
        }

并建立 _errorrecovery 队列。每个事件还有一个关联的 IConsumer，专门用于处理失败的事件。

 var subName = NameHelper.GetSubscriptionName(@namespace, _serviceName);

                        busConfig.ReceiveEndpoint(subName + "_errorrecovery", config =>
                        {
                            config.ConfigureConsumer(_simpleContainer, faultConsumers.Select(i => i.GenericType).ToArray());
                        });

这会产生一个名为 subname_errorrecovery 的队列和一个以事件命名的主题。该服务在主题中有订阅，但 _errorrecovery 也是如此。因此，每次向 Topic 发送消息时，事件的消费者和错误的消费者都会收到消息。

所以我正在寻找一种方法，将服务连接到恢复队列以及多个主题，而该队列也不会订阅每个主题。

我也很有可能以错误的方式处理这个问题。也许我只需要添加对重复消息的检查并将消息移动到主题，允许最初成功处理消息的服务简单地忽略它。这是我无论如何都打算做的事情，但我希望得到一些关于 MassTransit 错误处理的指导。

任何帮助将不胜感激。

【问题讨论】：

您是否需要使用订阅端点，或者您可以将您的消费者放在接收端点上并让 MassTransit 配置订阅以转发到您的接收端点？这是使用 MassTransit 发布/订阅的默认方法。

标签： azureservicebus masstransit azure-servicebus-queues azure-servicebus-topics

【解决方案1】：

Azure 服务总线允许您为每个为主题创建的订阅设置订阅过滤规则（请参阅https://docs.microsoft.com/en-us/azure/service-bus-messaging/topic-filters）。

您可以使用它为订阅定义过滤规则，以便只有与特定规则相对应的消息才会被放入该订阅者的队列进行处理。

这些过滤规则可以通过代码、ARM 模板、Azure 门户或 Azure CLI 设置。

您可以做的是确保在处理失败后将某个消息属性添加到消息中，然后再将其移至_error队列。我不太了解MassTransit，但我想会有一个选择。或者，MassTransit 本身可能已经提供了此类来源信息，以便您可以从设置的消息属性中得知消息在移动到 _error 队列之前在哪个订阅源（队列）进行了处理。

因此，让我们考虑一下何时将您的失败消息（可能是因为临时无法访问消费者服务的数据库而出现异常而失败）移动到 _error 队列，其中一些消息属性指示它已处理的位置。让我们将此属性称为“ErrorOrigin”，并为它指定一个唯一值来标识某个订阅。

因此，在您的情况下，您可以为每个订阅定义一个过滤规则，以便只有没有名为“ErrorOrigin”的属性的消息，或者如果设置了属性，则值必须匹配对应于该订阅的标识符名称（例如“subscriptionXforEventZ”）

如果您正确设置订阅过滤规则，您的订阅消费者现在将处理最初发布到该主题的所有消息，以及来自 _error 队列的也已发布到该主题的消息。但是，如果消息的“ErrorOrigin”属性对应于该订阅，Azure 服务总线只会将这些失败的消息放在相应的订阅队列中。

我已经在微服务架构的实践中成功地使用了这个特性。它允许向主题添加其他订阅者非常容易，但让这些订阅者可以选择仅接收来自主题的感兴趣的消息。

您可以在此处查看 Azure 服务总线上的订阅筛选示例，以更好地了解其工作原理： https://github.com/Azure/azure-service-bus/tree/master/samples/DotNet/Microsoft.ServiceBus.Messaging/TopicFilters

我希望这个想法可以帮助您解决您的具体问题。

【讨论】：

【解决方案2】：

此问题的开箱即用解决方案是添加主题订阅规则。因此，每当您要重新提交消息时，它只会根据订阅上应用的过滤器进入那些订阅。通过文章here了解主题订阅规则。

【讨论】：