c# - 调用所有处理程序后如何提交 UoW

Question

我有一个场景，其中单个节点上的所有处理程序都应该在调用所有处理程序后提交的单个“工作单元”上运行。我认为最好的方法是执行以下操作:

收到消息后，将这些操作作为管道的一部分执行:

1. 创建新的 DbContext 实例 (UoW)
2. 调用处理程序并传递 DbContext 实例
3. 如果调用所有处理程序而没有错误调用 DbContext.SaveChanges
4. 处理 DbContext

你能给我提示如何定制 Rebus 管道以满足上述要求吗？

编辑:

我已经结束了:

private static IBus _bus;

public void ConfigureServices(IServiceCollection services)
{
    services.AddDbContext<MyDbContext>();
    services.AddMvc();
    services.AddTransient<IBus>(sp => _bus);
}

public void Configure(IApplicationBuilder app, IHostingEnvironment env)
{
   StartRebus(app);
   ...
}

public static void StartRebus(this IApplicationBuilder app)
{
    var rebusServices = new ServiceCollection();
    rebusServices.AutoRegisterHandlersFromAssemblyOf<ActivityDenormalizer>();
    rebusServices.AddTransient<MyDbContext>(sp =>
    {
        var messageContext = MessageContext.Current
            ?? throw new InvalidOperationException("MessageContext.Current is null.");

        return messageContext.TransactionContext.Items
            .GetOrThrow<MyDbContext>(nameof(MyDbContext));
    });


    rebusServices.AddRebus((configure, sp) => configure
        .Transport(t => t.UseInMemoryTransport(new InMemNetwork(), "Messages"))
        .Options(o =>
        {
            o.EnableUnitOfWork<MyDbContext>(
                unitOfWorkFactoryMethod: messageContext =>
                {
                    //create new dbcontext instance regardless of ServiceLifeTime.
                    //Notice that I'm using ApplicationServices here, not RebusServices.
                    var dbContext = ActivatorUtilities.CreateInstance<MyDbContext>(app.ApplicationServices);
                    messageContext.TransactionContext.Items[nameof(MyDbContext)] = dbContext;
                    return dbContext;
                },
                commitAction: (messageContext, dbContext) => dbContext.SaveChanges(),
                cleanupAction: (messageContext, dbContext) => dbContext.Dispose());

        }));

    var rebusServiceProvider = rebusServices.BuildServiceProvider();
    rebusServiceProvider.UseRebus();
    _bus = rebusServiceProvider.GetRequiredService<IBus>();
}

应用程序服务和 rebus 服务在两个地方互连:

1. IBus 由 rebusServiceProvider 解析，但该实例也在应用程序服务中注册，因此我可以从我的应用程序向它发送消息。

2. MyDbContext 依赖项(DbContextOptions 等)由 ApplicationServices 解析，但 DbContext 也在 Rebus 的 unitOfWorkFactoryMethod 中实例化并在 rebusServices 中注册，以便它可以注入(inject) Rebus 处理程序。

Answer 1

是的——查看Rebus.UnitOfWork – 它有您需要的 Hook 。

具体来说，对于 Entity Framework，您可以执行如下操作:

Configure.With(new CastleWindsorContainerAdapter(container))
    .Transport(t => t.UseMsmq("test"))
    .Options(o => o.EnableUnitOfWork(
        async context => new CustomUnitOfWork(context, connectionString),
        commitAction: async (context, uow) => await uow.Commit()
    ))
    .Start();

CustomUnitOfWork 看起来像这样:

class CustomUnitOfWork
{
    public const string ItemsKey = "current-db-context";

    readonly MyDbContext _dbContext;

    public CustomUnitOfWork(IMessageContext messageContext, string connectionString)
    {
        _dbContext = new MyDbContext(connectionString);
        messageContext.TransactionContext.Items[ItemsKey] = this;
    }

    public async Task Commit()
    {
        await _dbContext.SaveChangesAsync();
    }

    public MyDbContext GetDbContext() => _dbContext;
}

然后您将设置您的 IoC 容器以通过从当前消息上下文中获取它来解析 MyDbContext – 对于 CaSTLe Windsor，这将像这样完成:

container.Register(
    Component.For<CustomUnitOfWork>()
        .UsingFactoryMethod(k =>
        {
            var messageContext = MessageContext.Current
                ?? throw new InvalidOperationException("Can't inject uow outside of Rebus handler");

            return messageContext.TransactionContext.Items
                .GetOrThrow<CustomUnitOfWork>(CustomUnitOfWork.ItemsKey);
        })
        .LifestyleTransient(),

    Component.For<MyDbContext>()
        .UsingFactoryMethod(k => k.Resolve<CustomUnitOfWork>().GetDbContext())
        .LifestyleTransient()
);