订阅自己发布的消息的 NServiceBus 主机

Question

NServiceBus 使用的版本：2.0.0.1145

问题：

是否可以配置 NServiceBus 主机，使其消费（订阅）自己发布的消息？

答案：

这似乎是可能的，但在以下配置中，在尝试将订阅插入订阅存储时，它给了我一个事务死锁异常。 当您使用 DbSubscriptionStorage 和超过 1 个“NumberOfWorkerThreads”时，就会发生这种情况。

错误：

无法执行命令：
INSERT INTO 订阅（SubscriberEndpoint、MessageType）值（@p0、@p1）
System.Data.SqlClinet.SqlException：
事务在锁资源上与另一个进程发生死锁，并被选为死锁受害者。重新运行事务。

之后，NServiceBus 尝试断开连接但失败，因为还有一个事务仍在进行中并抛出 UnhandledException。

如何重现：

这是我的 App.Config：

<!-- Publishing Configuration -->
<MsmqTransportConfig InputQueue="test_publisher_output" ErrorQueue="test_error" NumberOfWorkerThreads="3" MaxRetries="5" />

<!-- Subscription Configuration -->
<UnicastBusConfig DistributorControlAddress="" DistributorDataAddress="" ForwardReceivedMessagesTo="">
    <MessageEndpointMappings>
        <add Messages="MessageAssembly" Endpoint="test_publisher_output" />
    </MessageEndpointMappings>
</UnicastBusConfig>

我的总线配置：

var bus = Configure.With()
    .Log4Net()
    .StructureMapBuilder(container)
    .XmlSerializer()
    .MsmqTransport()
        .IsTransactional(true)
        .PurgeOnStartup(false)
    .DBSubcriptionStorage(subscriptionDbProperties, true)
    .Sagas()
    .NHibernateSagaPersister(sagaDbProperties, true)
    .UnicastBus()
        .ImpersonateSender(false)
        .LoadMessageHandlers(First<GridInterceptingMessageHandler>
            .Then<SagaMessageHandler>())
    .CreateBus()
    .Start();

这是我的订阅和 saga db 的 dbProperties：

connection.provider      NHibernate.Connection.DriverConnectionProvider
connection.driver_class  NHibernate.Driver.SqlClientDriver
dialect                  NHibernate.Dialect.MsSql2005Dialect

只要我不将 NumberOfWorkerThreads 增加到 1 以上，一切都会正常。一切都高于该值它会抛出上述错误。

我希望我没有忘记任何事情。提前感谢您的帮助。

Answer 1

如果您希望同一进程处理已发布的消息，最好在 Bus.Publish() 之后执行 Bus.SendLocal()。 SendLocal() 方法将在本地队列上放置一条消息，并且您的内部处理程序将拾取该消息并进行处理。这将消除你的僵局，但保持相同的语义。

Answer 2

我真的会考虑重新设计这个组件。如果您想要 nservicebus 为您提供的稳定性，并且您已经分解了组件，因此处理的每个部分都在单独的消息处理程序中，请将每个消息处理程序放入具有单独队列的单独可执行文件中。如果这是不可能的，那么您还没有真正获得 nservicebus 的稳定性，因为您被其他一些代码片段锁定，在这种情况下您应该直接调用所需的函数。

如果您只是测试它们全部从一个队列运行，那么在测试时也将它们分开。实际上没有理由订阅您自己的消息 - 如果可能的话，要么将处理程序拆分为单独的端点，如果不可能，则直接调用函数。