我希望批量处理相关消息,例如使用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中的数据存储的事务管理。
最佳答案
听起来像您钉了它:)
如果您得到正确的提交/回退行为,那么我会说这很好并且很花哨。
如果您对替代方法感兴趣,则可能想看看PerTransportMessage
Castle Windsor范围访问器-可以这样使用:
container.Register(
Component
.For<ISomething>()
.ImplementedBy<Whatever>()
.LifestyleScoped<PerTransportMessage>()
);
这应该能够实现完全相同的行为。
关于c# - 工作单元中多个消息的相同Rebus处理程序实例,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/15615961/