Rebus MSMQ 集中存储

Question

使用 MSMQ 启用集中式存储时，还有什么需要更改的吗？这是我所拥有的，但代码不会完全加载。如果我禁用集中式存储，将按预期工作。

_messageActivator = new BuiltinHandlerActivator();
_messageActivator.Register<PosOnlineHandler>(() => new PosOnlineHandler(WriteOutputAsync));
_messageActivator.Register<PumpDownHandler>(() => new PumpDownHandler(WriteOutputAsync));
_messageActivator.Register<MetersRequestHandler>(() => new MetersRequestHandler(WriteOutputAsync, _messageActivator.Bus));
_messageActivator.Register<CreditAuthorizationHandler>(() => new CreditAuthorizationHandler(WriteOutputAsync, _messageActivator.Bus));
Configure.With(_messageActivator)
   .Transport(t => t.UseMsmq("consumerServiceQueue"))
   .Routing(r => r.TypeBased()
      .Map<PumpDownEvent>("publisherServiceQueue")
      .Map<PosOnlineEvent>("publisherServiceQueue")
      .Map<MetersResponse>("publisherServiceQueue")
      .Map<CreditAuthorizationResponse>("publisherServiceQueue"))
   .Subscriptions(s => s.StoreInSqlServer(@"Data Source=(localdb)\MSSQLLocalDB;Initial Catalog=test;Integrated Security=True;Connect Timeout=30;Encrypt=False;TrustServerCertificate=True;ApplicationIntent=ReadWrite;MultiSubnetFailover=False", "RebusSubscriptions", true, true))
   .Start();
   _messageActivator.Bus.Subscribe<PumpDownEvent>().Wait();
   _messageActivator.Bus.Subscribe<PosOnlineEvent>().Wait();

谢谢。 斯科特 C.

Answer 1

这里有一个解释：Windows 窗体（您提到您从中调用此代码）使用坚持在 UI 线程上运行延续的任务调度程序。

这在 Windows 窗体上下文中非常有用，因为控件只能从该线程进行操作。

当您在Task 上调用.Wait()（或.Result）时会出现问题，因为这会阻塞当前线程，等待所有继续执行完成——但由于当前线程被阻塞，它永远不会可用于运行延续，从而导致死锁。

这里是another question that discusses this issue。正如您所看到的，这个问题绝不是 Rebus 特有的，它只是 async Tasks 和 Windows 窗体（或 WPF 或 ASP.NET）混合的一般特征。

我建议您在System.Windows.Forms.Form 的初始化期间调查如何正确运行async 代码，以便您可以通过awaitTasks 以推荐的方式建立您的订阅：

await bus.Subscribe<PumpDownEvent>();

await bus.Subscribe<PosOnlineEvent>();

---

为什么这个问题只在isCentralized: true 传递给StoreInSqlServer 配置方法时才会出现？

简单地说，这是因为当 Rebus 被配置为使用分散订阅存储和 MSMQ 作为传输时，实际上并没有发生真正的异步。

当运行集中式时，订阅者将自己注册为订阅存储中的订阅者。使用您使用的配置，这意味着SqlConnection 被（异步）打开，然后ExecuteNonQueryAsync（再次异步）执行，在[RebusSubscriptions] 表中插入一行。

运行去中心化时，订阅者通过向发布者发送SubscribeRequest 将自己注册为订阅者。 使用 MSMQ，这些操作恰好同步执行，而没有 awaiting 任何事情，因为 MSMQ 没有返回 Tasks 的 API。

如果您使用 SQL Server、Azure 服务总线、Amazon SQS 等作为传输，发送SubscribeRequest 时会出现一些异步内容，然后您会遇到同样的死锁。

我希望能解释它:)