NServiceBus：如何让订阅者订阅多个消息类型（每个来自不同的队列）答案

【问题标题】：NServiceBus: How to get a subscriber to subscribe to more than one message types (each from a different queue)NServiceBus：如何让订阅者订阅多个消息类型（每个来自不同的队列）
【发布时间】：2010-12-03 20:34:59
【问题描述】：

我希望能够让我的订阅者处理两个不同的消息流。我预计每种消息类型都会有一个 MSMQ 队列，但我看不到如何在我的 .config 文件的 MsmqTransportConfig 部分中指定多个 InputQueue。

这是我的订阅者的非 XML 配置：

        Configure.With(new[] { typeof(EventMessage), typeof(EventMessageHandler), typeof(NServiceBus.Unicast.Transport.CompletionMessage) })
            .CustomConfigurationSource(new UserConfigurationSource()
               .Register(() => new MsmqTransportConfig { InputQueue = "Subscriber1InputQueue", ErrorQueue = "error", NumberOfWorkerThreads = 1, MaxRetries = 5 }))
            .DefaultBuilder()
            .XmlSerializer()
            .MsmqTransport()
              .IsTransactional(true)
          .UnicastBus()
              .DoNotAutoSubscribe()
              .LoadMessageHandlers()
          .CreateBus()
          .Start();

编辑：我似乎从不同的人那里得到了不同的答案。谢谢大家！我想我有我的问题的答案，那就是：使用 NServiceBus 的进程（无论是发布者还是订阅者）只能在 SINGLE 队列上接收消息。对我来说，这是一个不必要的限制，不幸的是 NServiceBus 以这种方式工作。我不想有多个进程来接收消息，也不想让它们都进入同一个队列。如果特定消息处理程序出现问题，我希望只看到该特定消息类型的错误队列大小增加。我认为它可以更好地了解系统中发生的事情。

【问题讨论】：

+1 - 这是一个很好的问题，也是我正在努力解决的问题。

标签： nservicebus

【解决方案1】：

还没有使用 no-xml 配置，但是使用配置文件它看起来像：

<MsmqTransportConfig InputQueue="WorkerQueueForCurrentService" ErrorQueue="ErrorQueue" NumberOfWorkerThreads="1" MaxRetries="5"/>

<UnicastBusConfig>
    <MessageEndpointMappings>
        <add Messages="AssemblyName1" Endpoint="PublisherQueue1" />
        <add Messages="AssemblyName2.Message1, AssemblyName2" Endpoint="PublisherQueue2" />
        <add Messages="AssemblyName2.Message3, AssemblyName2" Endpoint="PublisherQueue2" />
    </MessageEndpointMappings>
</UnicastBusConfig>

因此，您当前服务的工作队列是“WorkerQueueForCurrentService”，它订阅在队列“PublisherQueue1”和“PublisherQueue2”上发布的不同消息。我已经包含了一个订阅整个消息程序集的示例（请参阅添加消息第 1 行）和给定消息程序集中的特定消息（请参阅添加消息第 2 行和第 3 行）。

Kristian kristenses 的回答不正确。输入队列与使用 nservicebus 的每个服务相关。无论是发布者还是订阅者。发布者在输入队列上接收订阅通知，订阅者将输入队列设置为发送给发布者的订阅通知的目标队列。

如果您想以编程方式订阅消息，例如 mrnye 所说，您需要一个消息端点映射。所以如果你做 bus.subscribe nservicebus 查看他的 messageendpointmappings 并尝试提取发布此消息的发布者队列名称。

messageendpointmappings 用于两者：
- 查找哪些消息在哪里发布
和
- 发送消息的目标队列，您使用 bus.send()

希望这可以解决一些问题:-)

【讨论】：

谢谢，hacktick。这确实使事情变得清晰。你说的一件事我认为我不同意：“messageendpointmappings 用于......查找哪些消息在哪里发布”。我认为这些实际上是进程想要订阅的端点/消息类型（无论是自动还是手动）。我认为这是因为我什至没有为我的发布者或订阅者服务定义这个，因为我不使用自动订阅机制，而是使用我自己的 ISubscriptionStorage 实现。
嗨 skb，“查找哪些消息在哪里发布”是指“如果我想订阅一条消息，我会执行 bus.subscribe(messagetype) 并且 nservicebus 会查找该消息的发布位置。以便 nservicebus 知道将订阅通知发送到哪里”。很抱歉定义不清楚。

【解决方案2】：

在 NServiceBus 中，所有消息都通过一个队列。它是队列：进程之间的 1:1 映射。因此，使用您的 DoNotAutoSubscribe()，您只需使用 app.config 中的映射手动订阅您想要的消息

例如，在配置后使用该功能进行订阅

_Bus.Subscribe<SomeMessage>();

抱歉，我不记得消息映射的语法了

【讨论】：

【解决方案3】：

看看Publish/Subscribe Configuration。 InputQueue 是为 Publisher 元素而不是为订阅者指定的。后者在 UnicastBusConfig 下的 MessageEndpointMappings 中添加它感兴趣的消息。如果您对两个不同的流感兴趣，只需在 MessageEndpointMappings 下添加元素。

【讨论】：

我真的很困惑。 :) 我一直在使用 MsmqTransportConfig 来指定我的 SUBSCRIBER 获取消息的方式......而且它有效！我已更新问题以反映我的配置。你能告诉我为什么会这样吗？
现在我还有一个问题。为什么 MessageEndPointMappings 元素上没有“ErrorQueue”值？似乎订阅者在处理消息时出现错误，它不知道将其放在哪里。我希望的是让“MessageQueueA”、“MessageQueueB”、“MessageQueueC”和“ErrorQueueA”、“ErrorQueueB”、“ErrorQueueC”都对应于消息类型“TypeA”、“TypeB”、“TypeC”，这样如果一条消息处理失败 5 次（因为订阅者的错误），我可以有另一个进程稍后将消息从“ErrorQueue*”移动到“MessageQueue*”。
MessageEndpointMappings 指定您的发布者所在的位置，因此您的订阅者知道将订阅消息发送到哪里。设置订阅后，发布者会将发布的消息发送到 InputQueue 中描述的端点（在您的示例中为“Subscriber1InputQueue”）。如果处理此消息失败，该消息将被移动到上面代码中指定的错误队列。这有意义吗？
Skb - 要实现单独的队列，您需要为每种消息类型单独处理。