【问题标题】:MassTransit saga running with prefetch > 1MassTransit 传奇以 prefetch > 1 运行
【发布时间】:2016-08-18 03:32:54
【问题描述】:

我有一个 MassTransit 传奇状态机(源自 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 块的 ThenAsync 处理程序被命中。此时,没有 saga 数据被持久化到存储库(如预期的那样)。
  2. StartupCompletedEvent 已发布到总线。这里也没有将 saga 数据持久化到 repo 中。
  3. Initially 声明的 ThenAsync 块完成。在此之后,saga 数据最终被持久化。
  4. 没有其他事情发生。

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

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

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

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

然后我执行 Console.WriteLine。我对此的理解是事件已被接收,但由于correlationId 不存在saga,因此被路由到Initially 处理程序。如果我在此时设置一个断点并检查 saga repo,则还没有 saga。

还有几点可能值得一提:

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

如果我遗漏了什么,请告诉我;我已经尝试提供我认为值得的所有内容,而不会让这个问题太长......

【问题讨论】:

  • 我发现了一个关于 MT3 的声明,声称事件会一直持续到持久化步骤 (groups.google.com/d/msg/masstransit-discuss/Jom6ns5jF-w/…)。但是,我根本没有看到这种行为。我正在使用 MT 3.3.5。
  • 如果你想推迟发布直到 saga 被持久化,你应该启用 InMemoryOutbox()。
  • @ChrisPatterson 谢谢 - 这似乎已经解决了。您能否详细说明为什么这不是默认设置?我无法想象你不会想要这种行为的情况......
  • 我知道从框架的角度来看它是一个中间件组件。但是从使用框架的开发人员的角度来看,我在这方面浪费了相当多的时间和精力;它在任何文档中都没有明确说明,在花了一些时间查看 MT 源代码之后,我不得不求助于一个 SO 问题。我仍然看不出在哪种情况下您实际上不想要这种行为。
  • 它仅用于发布其他消息并执行一些其他(事务性)事情(例如持久化)的消费者。但是中间件本身是在管道中启用的。因此,默认设置它会导致管道中的中间件无用。我绝对同意的是它需要在文档中。

标签: c# masstransit saga


【解决方案1】:

我也遇到了这个问题,我想发布 Chris 的评论作为答案,以便人们找到它。

解决方案是启用发件箱,以便保留消息直到 saga 被持久化。

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

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2017-10-19
    相关资源
    最近更新 更多