MassTransit 中的单身传奇答案

【问题标题】：Singleton saga in MassTransitMassTransit 中的单身传奇
【发布时间】：2021-08-19 03:22:07
【问题描述】：

我要做的是只创建一个 saga 实例，而不是由后续事件更新。例如，如果我在 09:00 发送 EventA，CorrelationId 等于 979158a2-dfa0-45f0-9107-8b8b8028bb9f，然后该事件的传奇在 10:00 仍然存在，那么如果在 10:00 我发送另一个 @ 实例987654324@ 与之前发送的 CorrelationId 相同，我不希望它覆盖之前的 saga 状态，而是希望 MT 忽略此事件。我有点实现了这一点，因为我的 saga 状态没有被覆盖（只有它的版本被增加），但问题是 MT 在新事件到达时抛出错误。例外是（+其上方的一个日志条目）：

[18:41:12 DBG] SAGA:App.Services.Communicators.Amazon.Dispatcher.Api.Sagas.GetOrdersState:0368a723-a819-450f-b6e5-9211d1d6a3a9 Dupe App.Services.Financial.Contracts.GetOrdersCommand
MongoDB.Driver.MongoWriteException: A write operation resulted in an error.
  E11000 duplicate key error collection: Dispatcher.get.orders.states index: _id_ dup key: { _id: BinData(3, 23A7680319A80F45B6E59211D1D6A3A9) }
 ---> MongoDB.Driver.MongoBulkWriteException`1[App.Services.Communicators.Amazon.Dispatcher.Api.Sagas.GetOrdersState]: A bulk write operation resulted in one or more errors.
  E11000 duplicate key error collection: Dispatcher.get.orders.states index: _id_ dup key: { _id: BinData(3, 23A7680319A80F45B6E59211D1D6A3A9) }
   at MongoDB.Driver.MongoCollectionImpl`1.BulkWriteAsync(IClientSessionHandle session, IEnumerable`1 requests, BulkWriteOptions options, CancellationToken cancellationToken)
   at MongoDB.Driver.MongoCollectionImpl`1.UsingImplicitSessionAsync[TResult](Func`2 funcAsync, CancellationToken cancellationToken)
   at MongoDB.Driver.MongoCollectionBase`1.InsertOneAsync(TDocument document, InsertOneOptions options, Func`3 bulkWriteAsync)
   --- End of inner exception stack trace ---
   at MongoDB.Driver.MongoCollectionBase`1.InsertOneAsync(TDocument document, InsertOneOptions options, Func`3 bulkWriteAsync)
   at MassTransit.MongoDbIntegration.Saga.Context.MongoDbSagaRepositoryContext`2.Insert(TSaga instance)

还有 saga 配置的重要部分：

Event(() => GetOrdersIntegrationEventReceived, e =>
    e
        .CorrelateById(x => x.Message.CorrelationId)
        .SelectId(x => x.Message.CorrelationId)
        .SetSagaFactory(ctx => new GetOrdersState
        {
            CorrelationId = ctx.Message.CorrelationId,
            PartnerId = ctx.Message.PartnerId,
            LastUpdatedBeforeFromRequest = ctx.Message.LastUpdatedBefore
        })
        .InsertOnInitial = true);

我正在尝试做的事情可能吗？

【问题讨论】：

标签： c# masstransit

【解决方案1】：

MassTransit 只会为每个CorrelationId 创建一个传奇实例，因此您的推理是正确的。但是，您的方法有点偏离，可以进行一些调整。

例如，您的事件配置：

Event(() => GetOrdersIntegrationEventReceived, e => e.CorrelateById(x => x.Message.CorrelationId));

就是这样，所有额外的东西都是您在调试日志中看到消息的原因。

那么，在声明事件之后：

During(Initial, Created,
    When(GetOrdersIntegrationEventReceived)
        .Then(context =>
        {
            context.Instance.PartnerId = context.Message.PartnerId,
            context.Instance.LastUpdatedBeforeFromRequest = context.Message.LastUpdatedBefore
        })
        .TransitionTo(Created)
);

就是这样。

【讨论】：