我希望批量处理相关消息,例如使用 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 的 Scoped 生活方式
_container.Register(
Component.For<IHandleMessages<CustomerCreated>, IHandleMessages<PreferredCustomer>>)
.ImplementedBy<CustomerHandler>().LifestyleScoped());
以及实例化 ScopedUnitOfWork 的自定义 UnitOfWorkManager
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 中的数据存储的事务管理。