gpt4 book ai didi

c# - 运行预取 > 1 的 MassTransit saga

转载 作者:行者123 更新时间:2023-11-30 23:22:36 27 4
gpt4 key购买 nike

我有一个 MassTransit saga 状态机(派生自 Automatonymous.MassTransitStateMachine),我正在尝试解决一个问题,该问题仅在我将端点配置 prefetchCount 设置为大于 1 的值时才会出现。

问题是“StartupCompletedEvent”已发布,然后在将 saga 状态持久保存到数据库之前立即进行处理。

状态机配置如下:

State(() => Initialising);
State(() => StartingUp);
State(() => GeneratingFiles);

Event(() => Requested, x => x.CorrelateById(ctx => ctx.Message.CorrelationId).SelectId(ctx => ctx.Message.CorrelationId));
Event(() => StartupCompleted, x => x.CorrelateById(ctx => ctx.Message.CorrelationId));
Event(() => InitialisationCompleted, x => x.CorrelateById(ctx => ctx.Message.CorrelationId));
Event(() => FileGenerationCompleted, x => x.CorrelateById(ctx => ctx.Message.CorrelationId));


Initially(
When(Requested)
.ThenAsync(async ctx =>
{
Console.WriteLine("Starting up...");
await ctx.Publish(new StartupCompletedEvent() { CorrelationId = ctx.Instance.CorrelationId }));
Console.WriteLine("Done starting up...");
}
.TransitionTo(StartingUp)
);


During(StartingUp,
When(StartupCompleted)
.ThenAsync(InitialiseSagaInstanceData)
.TransitionTo(Initialising)
);

// snip...!

当我的 saga 收到 Requested 事件时会发生什么:

  1. Initially block 的 ThenAsync 处理程序被命中。此时,没有 saga 数据保存到 repo(如预期的那样)。
  2. StartupCompletedEvent 发布到总线。这里的 repo 协议(protocol)中也没有保存传奇数据。
  3. Initially 声明的 ThenAsync block 完成。在此之后,saga 数据终于持久化了。
  4. 没有其他事情发生。

此时队列中没有消息,StartupCompletedEvent丢失。但是,数据库中有一个saga实例。

我已经尝试启动并确定其他线程之一(因为我的预取> 1)已经拾取了事件,没有在数据库中找到任何具有 correlationId 的传奇,并丢弃了事件。因此,事件在 saga 有机会被持久化之前被发布和处理。

如果我将以下内容添加到 Initially 处理程序中:

When(StartupCompleted)
.Then(ctx => Console.WriteLine("Got the startup completed event when there is no saga instance"))

然后我开始执行 Console.WriteLine。我对此的理解是事件已收到,但已路由到 Initially 处理程序,因为 correlationId 不存在任何传奇。如果我在此时放置一个断点并检查 saga repo 协议(protocol),则还没有 saga。

可能值得一提的其他几点:

  1. 我将 saga 存储库上下文设置为使用 IsolationLevel.Serializable
  2. 我正在使用 EntityFrameworkSagaRepository
  3. 当预取计数设置为 1 时,一切都按预期工作
  4. 我将 Ninject 用于 DI,并且我的 SagaRepository 是线程范围的,所以我想预取计数允许的每个处理程序都有自己的 saga 存储库副本
  5. 如果我在一个单独的线程中发布 StartupCompletedEvent,并且在它之前有 1000 毫秒的 sleep 时间,那么一切都会正常进行。我认为这是因为 saga 存储库已完成保留 saga 状态,因此当事件最终发布并由处理程序拾取时,将从存储库中正确检索 saga 状态。

如果我遗漏了什么,请告诉我;我试图提供我认为有值(value)的一切,而不让这个问题太长......

最佳答案

我也有这个问题,我想张贴 Chris 的评论作为答案,以便人们可以找到它。

解决方案是启用发件箱,以便在 saga 持久化之前保留消息。

c.ReceiveEndpoint("queue", e =>
{
e.UseInMemoryOutbox();
// other endpoint configuration here
}

关于c# - 运行预取 > 1 的 MassTransit saga,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/38716143/

27 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com