工作单元内多条消息的相同 Rebus 处理程序实例答案

【问题标题】：Same Rebus handler instance for multiple messages within a Unit of Work工作单元内多条消息的相同 Rebus 处理程序实例
【发布时间】：2023-03-09 20:25:02
【问题描述】：

我希望批量处理相关消息，例如使用 Rebus 服务总线在相同范围/事务中使用相同的处理程序（相同的实例）处理事件 CustomerCreated 和 PreferredCustomer。

同一个处理程序同时处理两个消息/事件：

class CustomerHandler : IHandleMessages<CustomerCreated>, IHandleMessages<PreferredCustomer>
{
    Customer Customer { get; set; }

    public CustomerHandler() {
        Customer = new Customer();
    }

    public void Handle(CustomerCreated message) {
        Customer.Name = message.Name;
        Console.WriteLine(Customer);
    }

    public void Handle(PreferredCustomer message) {
        Customer.Rebate = message.Rebate;
        Console.WriteLine(Customer);
    }
}

发送消息时我使用批处理操作（在 NServiceBus 中传输消息）

bus.Advanced.Batch.Publish(
  new CustomerCreated() { Name = "Anders" }, 
  new PreferredCustomer() { Rebate = 10 });

为了控制处理程序的生命周期，我使用Windsor Castle’s Scoped lifestyle

_container.Register(
  Component.For<IHandleMessages<CustomerCreated>, IHandleMessages<PreferredCustomer>>)
    .ImplementedBy<CustomerHandler>().LifestyleScoped());

还有一个自定义的 UnitOfWorkManager 实例化 ScopedUnitOfWork

class CustomUnitOfWorkManager : IUnitOfWorkManager
{
    private readonly IWindsorContainer _container;

    public CustomUnitOfWorkManager(IWindsorContainer container) {
        _container = container;
    }

    public IUnitOfWork Create() {
        return new ScopedUnitOfWork(_container);
    }
}

class ScopedUnitOfWork : IUnitOfWork
{
    private readonly IDisposable _scope;

    public ScopedUnitOfWork(IWindsorContainer container) {
        // Begin transaction
        _scope = container.BeginScope();

    }

    public void Dispose() {
        _scope.Dispose();
    }

    public void Commit() {
        // Commit transaction
        Console.WriteLine("Commiting");
    }

    public void Abort() {
        // Rollback transaction            
        Console.WriteLine("Aborting!!!");
    }
}

最终将 Rebus 配置为使用 CustomUnitOfWorkManager

var bus = Configure.With(new WindsorContainerAdapter(_container))
  .Transport(t => t.UseMsmqAndGetInputQueueNameFromAppConfig())
  .MessageOwnership(d => d.FromRebusConfigurationSection())
  .Events(x => x.AddUnitOfWorkManager(new CustomUnitOfWorkManager(_container)))
  .CreateBus()
  .Start();

这是正确的方法吗？

我有限的测试表明这应该有效。我什至可以扩展它以包括针对 ScopedUnitOfWork 中的数据存储的事务管理。

【问题讨论】：

标签： c# rebus

【解决方案1】：

听起来你已经搞定了:)

如果你得到正确的提交/回滚行为，那么我会说它很好而且花花公子。

如果您对替代方案感兴趣，您可能想看看PerTransportMessage Castle Windsor 范围访问器 - 它可以像这样使用：

container.Register(
    Component
        .For<ISomething>()
        .ImplementedBy<Whatever>()
        .LifestyleScoped<PerTransportMessage>()
);

应该能够实现完全相同的行为。

【讨论】：