使用 EFCore 持久性的自动匿名的多状态机不起作用
Multiple State machines with automatonymous using EFCore persistence not working
我们正在使用具有自动命名和 InMemoryRepository 的 Masstransit 来实现 saga 持久性。我们配置了大约 3 个状态机并且运行良好。我们最近从 InMemoryRepository 更改为 EFCore 以实现持久性。这导致只有第一个配置的状态机可以完美运行。其余的所有状态机甚至都没有进入 Initially 事件。需要帮助以了解实施是否正确。以下是代码详情:
公共交通状态机设置
services.AddMassTransit(x =>
{
x.AddBus(provider => MassTransit.Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.Host(hostUri, hst =>
{
hst.Username(appSettings.RabbitMQ.Username);
hst.Password(appSettings.RabbitMQ.Password);
});
cfg.ReceiveEndpoint("microservice-response", e =>
{
AddConsumers(e, provider);
e.ConfigureSaga<ServiceRequest1RegisterState>(provider);
e.ConfigureSaga<ServiceRequest1UpdateState>(provider);
e.ConfigureSaga<ServiceRequest1ApproveState>(provider);
});
}));
////x.AddSagaStateMachine<ServiceRequest1RegisterStateMachine, ServiceRequest1RegisterState>()
//// .InMemoryRepository();
////x.AddSagaStateMachine<ServiceRequest1UpdateStateMachine, ServiceRequest1UpdateState>()
//// .InMemoryRepository();
////x.AddSagaStateMachine<ServiceRequest1ApproveStateMachine, ServiceRequest1ApproveState>()
//// .InMemoryRepository();
x.AddSagaStateMachine<ServiceRequest1RegisterStateMachine, ServiceRequest1RegisterState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.AddDbContext<DbContext, ServiceRequest1RegisterStateDbContext>((provider, builder) =>
{
builder.UseSqlServer(configuration.GetConnectionString("StateDBConnection"), m =>
{
m.MigrationsAssembly(Assembly.GetExecutingAssembly().GetName().Name);
m.MigrationsHistoryTable($"__{nameof(ServiceRequest1RegisterStateDbContext)}");
});
});
});
x.AddSagaStateMachine<ServiceRequest1UpdateStateMachine, ServiceRequest1UpdateState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.AddDbContext<DbContext, ServiceRequest1UpdateStateDbContext>((provider, builder) =>
{
builder.UseSqlServer(configuration.GetConnectionString("StateDBConnection"), m =>
{
m.MigrationsAssembly(Assembly.GetExecutingAssembly().GetName().Name);
m.MigrationsHistoryTable($"__{nameof(ServiceRequest1UpdateStateDbContext)}");
});
});
});
x.AddSagaStateMachine<ServiceRequest1ApproveStateMachine, ServiceRequest1ApproveState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.AddDbContext<DbContext, ServiceRequest1ApproveStateDbContext>((provider, builder) =>
{
builder.UseSqlServer(configuration.GetConnectionString("StateDBConnection"), m =>
{
m.MigrationsAssembly(Assembly.GetExecutingAssembly().GetName().Name);
m.MigrationsHistoryTable($"__{nameof(ServiceRequest1ApproveStateDbContext)}");
});
});
});
});
services.AddSingleton<IHostedService, MassTransitBusService>();
所有 3 个状态机的 DBContext 如下所示
public class ServiceRequest1ApproveStateDbContext : SagaDbContext
{
public ServiceRequest1ApproveStateDbContext(DbContextOptions<ServiceRequest1ApproveStateDbContext> options)
: base(options)
{
}
/// <summary>
/// Gets the configurations.
/// </summary>
protected override IEnumerable<ISagaClassMap> Configurations
{
get { yield return new ServiceRequest1ApproveStateMap(); }
}
}
状态图如下
public class ServiceRequest1ApproveStateMap : SagaClassMap<ServiceRequest1ApproveState>
{
protected override void Configure(EntityTypeBuilder<ServiceRequest1ApproveState> entity, ModelBuilder model)
{
entity.Property(x => x.CurrentState).HasMaxLength(64);
entity.Property(x => x.Id);
entity.Property(x => x.ServiceId);
entity.Property(x => x.ReadyEventStatus);
}
}
从上面的代码来看,只有ServiceRequest1RegisterStateMachine可以正常工作,其余状态机甚至都没有进入Initially。我可以确认迁移都是事先完成的,并且当只有 1 个存储库配置存在时,这些状态机都可以单独工作,即,如果只有 ServiceRequest1ApproveStateMachine 的回购配置,这个状态机工作正常。但是如果所有 3 个 repo 配置都存在,那么只有第一个有效。
我需要有关使用 EFCore 正确实施 saga 持久性的指导。我应该尝试使用 masstransit 文档中提到的单个 dbcontext 来实现吗(https://masstransit-project.com/usage/sagas/efcore.html)
下面是我为确保状态机按预期工作所做的更改。这些更改是根据@Chris 在评论中的建议完成的,尽管我们最初对状态机的实现是基于 sqlserver efcore 持久性上的文档。
services.AddDbContext<ServiceRequest1RegisterStateDbContext>(x =>
x.UseSqlServer(configuration.GetConnectionString("StateDBConnection")));
services.AddDbContext<ServiceRequest1UpdateStateDbContext>(x =>
x.UseSqlServer(configuration.GetConnectionString("StateDBConnection")));
services.AddDbContext<ServiceRequest1ApproveStateDbContext>(x =>
x.UseSqlServer(configuration.GetConnectionString("StateDBConnection")));
services.AddMassTransit(x =>
{
x.AddBus(provider => MassTransit.Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.Host(hostUri, hst =>
{
hst.Username(appSettings.RabbitMQ.Username);
hst.Password(appSettings.RabbitMQ.Password);
});
cfg.ReceiveEndpoint("microservice-response", e =>
{
AddConsumers(e, provider);
e.ConfigureSaga<ServiceRequest1RegisterState>(provider);
e.ConfigureSaga<ServiceRequest1UpdateState>(provider);
e.ConfigureSaga<ServiceRequest1ApproveState>(provider);
});
}));
////x.AddSagaStateMachine<ServiceRequest1RegisterStateMachine, ServiceRequest1RegisterState>()
//// .InMemoryRepository();
////x.AddSagaStateMachine<ServiceRequest1UpdateStateMachine, ServiceRequest1UpdateState>()
//// .InMemoryRepository();
////x.AddSagaStateMachine<ServiceRequest1ApproveStateMachine, ServiceRequest1ApproveState>()
//// .InMemoryRepository();
x.AddSagaStateMachine<ServiceRequest1RegisterStateMachine, ServiceRequest1RegisterState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.ExistingDbContext<ServiceRequest1RegisterStateDbContext>();
});
x.AddSagaStateMachine<ServiceRequest1UpdateStateMachine, ServiceRequest1UpdateState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.ExistingDbContext<ServiceRequest1UpdateStateDbContext>();
});
x.AddSagaStateMachine<ServiceRequest1ApproveStateMachine, ServiceRequest1ApproveState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.ExistingDbContext<ServiceRequest1ApproveStateDbContext>();
});
});
services.AddSingleton<IHostedService, MassTransitBusService>();
}
我们正在使用具有自动命名和 InMemoryRepository 的 Masstransit 来实现 saga 持久性。我们配置了大约 3 个状态机并且运行良好。我们最近从 InMemoryRepository 更改为 EFCore 以实现持久性。这导致只有第一个配置的状态机可以完美运行。其余的所有状态机甚至都没有进入 Initially 事件。需要帮助以了解实施是否正确。以下是代码详情:
公共交通状态机设置
services.AddMassTransit(x =>
{
x.AddBus(provider => MassTransit.Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.Host(hostUri, hst =>
{
hst.Username(appSettings.RabbitMQ.Username);
hst.Password(appSettings.RabbitMQ.Password);
});
cfg.ReceiveEndpoint("microservice-response", e =>
{
AddConsumers(e, provider);
e.ConfigureSaga<ServiceRequest1RegisterState>(provider);
e.ConfigureSaga<ServiceRequest1UpdateState>(provider);
e.ConfigureSaga<ServiceRequest1ApproveState>(provider);
});
}));
////x.AddSagaStateMachine<ServiceRequest1RegisterStateMachine, ServiceRequest1RegisterState>()
//// .InMemoryRepository();
////x.AddSagaStateMachine<ServiceRequest1UpdateStateMachine, ServiceRequest1UpdateState>()
//// .InMemoryRepository();
////x.AddSagaStateMachine<ServiceRequest1ApproveStateMachine, ServiceRequest1ApproveState>()
//// .InMemoryRepository();
x.AddSagaStateMachine<ServiceRequest1RegisterStateMachine, ServiceRequest1RegisterState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.AddDbContext<DbContext, ServiceRequest1RegisterStateDbContext>((provider, builder) =>
{
builder.UseSqlServer(configuration.GetConnectionString("StateDBConnection"), m =>
{
m.MigrationsAssembly(Assembly.GetExecutingAssembly().GetName().Name);
m.MigrationsHistoryTable($"__{nameof(ServiceRequest1RegisterStateDbContext)}");
});
});
});
x.AddSagaStateMachine<ServiceRequest1UpdateStateMachine, ServiceRequest1UpdateState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.AddDbContext<DbContext, ServiceRequest1UpdateStateDbContext>((provider, builder) =>
{
builder.UseSqlServer(configuration.GetConnectionString("StateDBConnection"), m =>
{
m.MigrationsAssembly(Assembly.GetExecutingAssembly().GetName().Name);
m.MigrationsHistoryTable($"__{nameof(ServiceRequest1UpdateStateDbContext)}");
});
});
});
x.AddSagaStateMachine<ServiceRequest1ApproveStateMachine, ServiceRequest1ApproveState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.AddDbContext<DbContext, ServiceRequest1ApproveStateDbContext>((provider, builder) =>
{
builder.UseSqlServer(configuration.GetConnectionString("StateDBConnection"), m =>
{
m.MigrationsAssembly(Assembly.GetExecutingAssembly().GetName().Name);
m.MigrationsHistoryTable($"__{nameof(ServiceRequest1ApproveStateDbContext)}");
});
});
});
});
services.AddSingleton<IHostedService, MassTransitBusService>();
所有 3 个状态机的 DBContext 如下所示
public class ServiceRequest1ApproveStateDbContext : SagaDbContext
{
public ServiceRequest1ApproveStateDbContext(DbContextOptions<ServiceRequest1ApproveStateDbContext> options)
: base(options)
{
}
/// <summary>
/// Gets the configurations.
/// </summary>
protected override IEnumerable<ISagaClassMap> Configurations
{
get { yield return new ServiceRequest1ApproveStateMap(); }
}
}
状态图如下
public class ServiceRequest1ApproveStateMap : SagaClassMap<ServiceRequest1ApproveState>
{
protected override void Configure(EntityTypeBuilder<ServiceRequest1ApproveState> entity, ModelBuilder model)
{
entity.Property(x => x.CurrentState).HasMaxLength(64);
entity.Property(x => x.Id);
entity.Property(x => x.ServiceId);
entity.Property(x => x.ReadyEventStatus);
}
}
从上面的代码来看,只有ServiceRequest1RegisterStateMachine可以正常工作,其余状态机甚至都没有进入Initially。我可以确认迁移都是事先完成的,并且当只有 1 个存储库配置存在时,这些状态机都可以单独工作,即,如果只有 ServiceRequest1ApproveStateMachine 的回购配置,这个状态机工作正常。但是如果所有 3 个 repo 配置都存在,那么只有第一个有效。 我需要有关使用 EFCore 正确实施 saga 持久性的指导。我应该尝试使用 masstransit 文档中提到的单个 dbcontext 来实现吗(https://masstransit-project.com/usage/sagas/efcore.html)
下面是我为确保状态机按预期工作所做的更改。这些更改是根据@Chris 在评论中的建议完成的,尽管我们最初对状态机的实现是基于 sqlserver efcore 持久性上的文档。
services.AddDbContext<ServiceRequest1RegisterStateDbContext>(x =>
x.UseSqlServer(configuration.GetConnectionString("StateDBConnection")));
services.AddDbContext<ServiceRequest1UpdateStateDbContext>(x =>
x.UseSqlServer(configuration.GetConnectionString("StateDBConnection")));
services.AddDbContext<ServiceRequest1ApproveStateDbContext>(x =>
x.UseSqlServer(configuration.GetConnectionString("StateDBConnection")));
services.AddMassTransit(x =>
{
x.AddBus(provider => MassTransit.Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.Host(hostUri, hst =>
{
hst.Username(appSettings.RabbitMQ.Username);
hst.Password(appSettings.RabbitMQ.Password);
});
cfg.ReceiveEndpoint("microservice-response", e =>
{
AddConsumers(e, provider);
e.ConfigureSaga<ServiceRequest1RegisterState>(provider);
e.ConfigureSaga<ServiceRequest1UpdateState>(provider);
e.ConfigureSaga<ServiceRequest1ApproveState>(provider);
});
}));
////x.AddSagaStateMachine<ServiceRequest1RegisterStateMachine, ServiceRequest1RegisterState>()
//// .InMemoryRepository();
////x.AddSagaStateMachine<ServiceRequest1UpdateStateMachine, ServiceRequest1UpdateState>()
//// .InMemoryRepository();
////x.AddSagaStateMachine<ServiceRequest1ApproveStateMachine, ServiceRequest1ApproveState>()
//// .InMemoryRepository();
x.AddSagaStateMachine<ServiceRequest1RegisterStateMachine, ServiceRequest1RegisterState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.ExistingDbContext<ServiceRequest1RegisterStateDbContext>();
});
x.AddSagaStateMachine<ServiceRequest1UpdateStateMachine, ServiceRequest1UpdateState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.ExistingDbContext<ServiceRequest1UpdateStateDbContext>();
});
x.AddSagaStateMachine<ServiceRequest1ApproveStateMachine, ServiceRequest1ApproveState>()
.EntityFrameworkRepository(r =>
{
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.ExistingDbContext<ServiceRequest1ApproveStateDbContext>();
});
});
services.AddSingleton<IHostedService, MassTransitBusService>();
}