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 Forms（你提到你从中调用这段代码）使用了一个任务调度程序，它坚持在 UI 线程上运行ning 继续。

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

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

这是another question that discusses this issue。如您所见，这个问题绝不是 Rebus 特有的，它只是异步 Tasks 和 Windows 表单（或 WPF 或 ASP.NET 混合的一般特征事）。

我建议您研究如何在 System.Windows.Forms.Form 的初始化期间正确运行 async 代码，这样您就可以按照 await 推荐的方式建立您的订阅正在 Tasks:

await bus.Subscribe<PumpDownEvent>();

await bus.Subscribe<PosOnlineEvent>();

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

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

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

当运行宁去中心化时，订阅者通过向发布者发送SubscribeRequest将自己注册为订阅者。 对于 MSMQ，此操作恰好同步执行，没有 await 任何事情，因为 MSMQ 没有 API returns [=13] =]s.

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

我希望解释清楚 :)

Rebus MSMQ 集中存储

Rebus MSMQ Centralized Storage

rebus