如何在实体框架中持久保存 MassTransit 状态数据？
Posted
技术标签:
【中文标题】如何在实体框架中持久保存 MassTransit 状态数据？【英文标题】：How can i persist MassTransit state data in entity framework? 【发布时间】：2021-05-01 20:12:32 【问题描述】：我有一个使用 MassTransit 的工作状态机，但是，我想使用 EF Core 持久化状态存储。我按照 MassTransit 网站上的指南进行操作，但它不起作用，它会生成表格，但在使用状态机时不会用数据填充表格。
这是我在 startup.cs 中的代码：
var saga = new OrderStateMachine();
var repo = new InMemorySagaRepository<OrderStateData>();
services.TryAddSingleton(KebabCaseEndpointNameFormatter.Instance);
services.AddMassTransit(config =>
config.AddSagaRepository<OrderStateData>()
.EntityFrameworkRepository(r =>
r.ExistingDbContext<OrderContext>();
);
config.AddSagaStateMachine<OrderStateMachine, OrderStateData>()
.EntityFrameworkRepository(r =>
r.ConcurrencyMode = ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.AddDbContext<DbContext, OrderContext>((provider,builder) =>
builder.Usemysql(connString, ServerVersion.AutoDetect(connString) , m =>
m.MigrationsAssembly(Assembly.GetExecutingAssembly().GetName().Name);
m.MigrationsHistoryTable($"__nameof(OrderContext)");
);
);
);
config.UsingRabbitMq((ctx, cfg) =>
cfg.Host(EnvironmentVariables.RabbitMqConnectionString);
cfg.ReceiveEndpoint("Constants.OrderBus1", c =>
c.StateMachineSaga(saga, repo);
);
);
);
services.AddMassTransitHostedService();
这是我的 EF 核心 DBContext：
public class OrderContext : SagaDbContext
public DbSet<Order> Orders get; set;
public OrderContext(DbContextOptions<OrderContext> options)
: base(options)
protected override void OnModelCreating(ModelBuilder modelBuilder)
protected override IEnumerable<ISagaClassMap> Configurations
get yield return new OrderStateMap();
最后是我的 orderstatemap
public class OrderStateMap :
SagaClassMap<OrderStateData>
protected override void Configure(EntityTypeBuilder<OrderStateData> entity, ModelBuilder model)
entity.Property(x => x.CurrentState).HasMaxLength(64);
entity.Property(x => x.OrderCreationDateTime);
entity.Property(x => x.OrderId);
entity.Property(x => x.PackagesId);
entity.Property(x => x.PaymentMethod);
entity.Property(x => x.CheckoutSessionId);
entity.Property(x => x.OrderCancelDateTime);
entity.Property(x => x.OrderCollectedDateTime);
我认为这与我的定义有关 " var repo = new InMemorySagaRepository();" 但我不知道用什么来代替它。
编辑：
应用修复后出现以下错误：
fail: Microsoft.EntityFrameworkCore.Database.Command[20102]
Failed executing DbCommand (270ms) [Parameters=[p0='?' (DbType = Guid)], CommandType='Text', CommandTimeout='30']
SELECT `o`.`CorrelationId`, `o`.`CheckoutSessionId`, `o`.`CurrentState`, `o`.`OrderCancelDateTime`, `o`.`OrderCollectedDateTime`, `o`.`OrderCreationDateTime`, `o`.`OrderId`, `o`.`PackagesId`, `o`.`PaymentMethod`
FROM (
SELECT * FROM dbo.OrderStateData WITH (UPDLOCK, ROWLOCK) WHERE CorrelationId = @p0
) AS `o`
LIMIT 2
fail: Microsoft.EntityFrameworkCore.Query[10100]
An exception occurred while iterating over the results of a query for context type 'OrderService.DAL.OrderContext'.
MySqlConnector.MySqlException (0x80004005): You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '(UPDLOCK, ROWLOCK) WHERE CorrelationId = '1198e061-d2d4-463f-9db6-117598
' at line 3
---> MySqlConnector.MySqlException (0x80004005): You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '(UPDLOCK, ROWLOCK) WHERE CorrelationId = '1198e061-d2d4-463f-9db6-
' at line 3c'
at MySqlConnector.Core.ServerSession.ReceiveReplyAsyncAwaited(ValueTask`1 task) in /_/src/MySqlConnector/Core/ServerSession.cs:line 815
at MySqlConnector.Core.ResultSet.ReadResultSetHeaderAsync(IOBehavior ioBehavior) in /_/src/MySqlConnector/Core/ResultSet.cs:line 49
at MySqlConnector.MySqlDataReader.ActivateResultSet(CancellationToken cancellationToken) in /_/src/MySqlConnector/MySqlDataReader.cs:line 131
at MySqlConnector.MySqlDataReader.CreateAsync(CommandListPosition commandListPosition, ICommandPayloadCreator payloadCreator, IDictionary`2 cachedProcedures, IMySqlCommand command, CommandBehavior behavior, IOBehavior ioBehavior, CancellationToken
cancellationToken) in /_/src/MySqlConnector/MySqlDataReader.cs:line 436
at MySqlConnector.Core.CommandExecutor.ExecuteReaderAsync(IReadOnlyList`1 commands, ICommandPayloadCreator payloadCreator, CommandBehavior behavior, IOBehavior ioBehavior, CancellationToken cancellationToken) in /_/src/MySqlConnector/Core/CommandEx
ecutor.cs:line 60
at MySqlConnector.MySqlCommand.ExecuteReaderAsync(CommandBehavior behavior, IOBehavior ioBehavior, CancellationToken cancellationToken) in /_/src/MySqlConnector/MySqlCommand.cs:line 310
at MySqlConnector.MySqlCommand.ExecuteDbDataReaderAsync(CommandBehavior behavior, CancellationToken cancellationToken) in /_/src/MySqlConnector/MySqlCommand.cs:line 304
at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
at Microsoft.EntityFrameworkCore.Query.Internal.SingleQueryingEnumerable`1.AsyncEnumerator.InitializeReaderAsync(DbContext _, Boolean result, CancellationToken cancellationToken)
at Pomelo.EntityFrameworkCore.MySql.Storage.Internal.MySqlExecutionStrategy.ExecuteAsync[TState,TResult](TState state, Func`4 operation, Func`4 verifySucceeded, CancellationToken cancellationToken)
at Microsoft.EntityFrameworkCore.Query.Internal.SingleQueryingEnumerable`1.AsyncEnumerator.MoveNextAsync()
MySqlConnector.MySqlException (0x80004005): You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '(UPDLOCK, ROWLOCK) WHERE CorrelationId = '1198e061-d2d4-463f-9db6-117598
' at line 3
---> MySqlConnector.MySqlException (0x80004005): You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '(UPDLOCK, ROWLOCK) WHERE CorrelationId = '1198e061-d2d4-463f-9db6-
' at line 3c'
at MySqlConnector.Core.ServerSession.ReceiveReplyAsyncAwaited(ValueTask`1 task) in /_/src/MySqlConnector/Core/ServerSession.cs:line 815
at MySqlConnector.Core.ResultSet.ReadResultSetHeaderAsync(IOBehavior ioBehavior) in /_/src/MySqlConnector/Core/ResultSet.cs:line 49
at MySqlConnector.MySqlDataReader.ActivateResultSet(CancellationToken cancellationToken) in /_/src/MySqlConnector/MySqlDataReader.cs:line 131
at MySqlConnector.MySqlDataReader.CreateAsync(CommandListPosition commandListPosition, ICommandPayloadCreator payloadCreator, IDictionary`2 cachedProcedures, IMySqlCommand command, CommandBehavior behavior, IOBehavior ioBehavior, CancellationToken
cancellationToken) in /_/src/MySqlConnector/MySqlDataReader.cs:line 436
at MySqlConnector.Core.CommandExecutor.ExecuteReaderAsync(IReadOnlyList`1 commands, ICommandPayloadCreator payloadCreator, CommandBehavior behavior, IOBehavior ioBehavior, CancellationToken cancellationToken) in /_/src/MySqlConnector/Core/CommandEx
ecutor.cs:line 60
at MySqlConnector.MySqlCommand.ExecuteReaderAsync(CommandBehavior behavior, IOBehavior ioBehavior, CancellationToken cancellationToken) in /_/src/MySqlConnector/MySqlCommand.cs:line 310
at MySqlConnector.MySqlCommand.ExecuteDbDataReaderAsync(CommandBehavior behavior, CancellationToken cancellationToken) in /_/src/MySqlConnector/MySqlCommand.cs:line 304
at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
at Microsoft.EntityFrameworkCore.Storage.RelationalCommand.ExecuteReaderAsync(RelationalCommandParameterObject parameterObject, CancellationToken cancellationToken)
at Microsoft.EntityFrameworkCore.Query.Internal.SingleQueryingEnumerable`1.AsyncEnumerator.InitializeReaderAsync(DbContext _, Boolean result, CancellationToken cancellationToken)
at Pomelo.EntityFrameworkCore.MySql.Storage.Internal.MySqlExecutionStrategy.ExecuteAsync[TState,TResult](TState state, Func`4 operation, Func`4 verifySucceeded, CancellationToken cancellationToken)
at Microsoft.EntityFrameworkCore.Query.Internal.SingleQueryingEnumerable`1.AsyncEnumerator.MoveNextAsync()
我尝试添加一个新的数据库迁移，但这并没有解决问题
【问题讨论】：
您需要为 MySQL 创建一个 LockStatementProvider，它将为 MySQL 正确格式化 lock 语句。 PostgreSQL 可以用作示例。我不确定 MySQL 的语法。 【参考方案1】：您的第一个配置脚本的正确版本如下所示。您定义了很多重复类型，我们正在正确配置接收端点。
services.AddMassTransit(config =>
config.SetKebabCaseEndpointNameFormatter();
config.AddSagaStateMachine<OrderStateMachine, OrderStateData>()
.EntityFrameworkRepository(r =>
r.ConcurrencyMode = ConcurrencyMode.Pessimistic;
// you will need to create the lock statement provider
r.LockStatementProvider = new MySqlLockStatementProvider();
r.AddDbContext<DbContext, OrderContext>((provider,builder) =>
builder.UseMySql(connString, ServerVersion.AutoDetect(connString) , m =>
m.MigrationsAssembly(Assembly.GetExecutingAssembly().GetName().Name);
m.MigrationsHistoryTable($"__nameof(OrderContext)");
);
);
);
config.UsingRabbitMq((ctx, cfg) =>
cfg.Host(EnvironmentVariables.RabbitMqConnectionString);
cfg.ReceiveEndpoint("Constants.OrderBus1", c =>
c.ConfigureSaga<OrderStateData>(ctx);
);
);
);
services.AddMassTransitHostedService();
【讨论】：
感谢您的评论！当我使用您的代码时，我收到一条错误消息，提示找不到“c.ConfigureSagaStateMachine”方法。我用 ConfigureSaga 替换了它，这让我可以编译。但是我现在得到一个与无效 SQL 语法相关的运行时错误我用错误更新了我的答案 嗨！我已经更新了我的代码并编写了我自己的 LockStatementProvider。它现在可以很好地写入我的数据库，购买我的状态机不再工作。当我删除 c.configureSage 并用我的旧 c.StateMachineSaga(saga, repo); 替换它时，它只会到达启动块并且不会继续。它再次起作用，但这显然不允许我使用数据库部分。【参考方案2】：我使用了 PostgreSQL 的下一个配置。
services.AddMassTransit(cfg =>
cfg.SetKebabCaseEndpointNameFormatter();
cfg.AddConsumersFromNamespaceContaining<SubmitOrderConsumer>();
cfg.AddSagaStateMachine<OrderStateMachine, OrderState>()
.EntityFrameworkRepository(r =>
r.ConcurrencyMode =
ConcurrencyMode.Pessimistic; // or use Optimistic, which requires RowVersion
r.LockStatementProvider = new PostgresLockStatementProvider();
r.AddDbContext<DbContext, OrderStateDbContext>((provider, builder) =>
builder.UseNpgsql(
hostContext.Configuration.GetConnectionString("DefaultConnection"), m =>
m.MigrationsAssembly(Assembly.GetExecutingAssembly().GetName().Name);
m.MigrationsHistoryTable($"__nameof (OrderStateDbContext)");
);
);
);
cfg.UsingRabbitMq(/*your config*/);
);
services.AddMassTransitHostedService();
)
【讨论】：
以上是关于如何在实体框架中持久保存 MassTransit 状态数据？的主要内容，如果未能解决你的问题，请参考以下文章
如何使用 managedObjectContext 来使用实体？
MassTransit一个优秀的.NET消息（事件）总线框架