【问题标题】：NServiceBus Host that Subscribes to his own Published Messages订阅自己发布的消息的 NServiceBus 主机
【发布时间】：2010-10-15 13:50:42
【问题描述】：

使用的 NServiceBus 版本：2.0.0.1145

问题：

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

答案：

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

错误：

Could not execute command: INSERT INTO Subscription (SubscriberEndpoint, MessageType) VALUES (@p0, @p1) System.Data.SqlClinet.SqlException: Transaction was deadlocked on lock resources with another process and has been chosen as the deadlock victim. Rerun the transaction.

之后，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();

这是我的订阅和传奇数据库的 dbProperties：

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

只要我不将 NumberOfWorkerThreads 增加到 1 以上，一切正常。高于 1 的所有内容都会引发上述错误。

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

【问题讨论】：

为什么同一个端点想要订阅它自己发布的消息？我的意思是，如果您想在同一个端点中发生其他事情，您可以将它们放在最初进行发布的代码中。
嗨，Udi，因为我会放松 nservicebus 给我的稳定性。我想将一大堆工作分解成由消息处理程序执行的小包。这里完成的工作示例是调用 web 服务、打开 ftp 连接、创建和转换 xml 文件、数据库事务等。我也希望可以选择仅发布或发送消息，并且不想关心端点。应用程序的管理员应该决定在哪里托管端点。在最简单的配置中，这一系列工作应该在一个主机中完成。

标签： nservicebus publish-subscribe database-deadlocks

【解决方案1】：

如果您希望相同的进程来处理已发布的消息，最好在 Bus.Publish() 之后执行 Bus.SendLocal()。 SendLocal() 方法将在本地队列上放置一条消息，您的内部处理程序将拾取并处理它。这将摆脱您的死锁，同时保持相同的语义。

【讨论】：

嗨亚当，没有问题是订阅而不是发送或发布消息。因此在启动没有发送或发布消息的主机时已经出现问题。

【解决方案2】：

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

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

【讨论】：

这种见解对我来说令人惊讶，这似乎是正确的答案。拆分您的服务。 nservicebus 仅在您的服务的外部边界上。了解发送、发布、回复的用法。 nservicebus 不支持我对上述问题的实现（按设计）。
为什么不呢？如果我的可执行文件有一个部分需要很长时间来处理，为什么不让该部分由命令启动并导致一个传奇完成它？ NServiceBus 将确定处理 server1 与我最初发送消息（给它自己）的 exe1 现在很忙，而另一个具有我相同 exe1 的服务器现在可以正确处理它！不？如果从外部客户端接收到两个相同的命令，并且我运行 exe1 的服务器陷入困境，这不是应该发生的情况吗？ NServiceBus 不会自动处理这个并将其发送到运行 exe1 的 server2 吗？
这似乎与 CQRS 模型相矛盾。您的查询端（在同一进程中）将具有将该消息转发到非规范化器的事件处理程序。