MassTransit Mediator(v7)如何在所有事件处理器中传递CorrelationId？

要让同一操作链中的所有处理器都能获取到初始的CorrelationId，核心是在发布后续事件时继承当前消费上下文的关联信息，具体实现步骤如下：

1. 确保事件正确实现CorrelatedBy<Guid>接口 #

虽然你已经添加了CorrelationId属性，但必须明确实现CorrelatedBy<Guid>接口，让MassTransit能够识别事件的关联ID字段：

public class CorrelatedTestEvent : CorrelatedBy<Guid>
{
    public Guid CorrelationId { get; set; }
}

public class CorrelatedTestEvent2 : CorrelatedBy<Guid>
{
    public Guid CorrelationId { get; set; }
}

public class CorrelatedTestEvent3 : CorrelatedBy<Guid>
{
    public Guid CorrelationId { get; set; }
}

2. 在处理器中使用ConsumeContext.Publish发布后续事件 #

不要直接使用IMediator.Publish，而是通过当前的ConsumeContext发布新事件，这样MassTransit会自动将当前上下文的CorrelationId传递给新事件的消费上下文：

public class CorrelatedTestEventConsumer : IConsumer<CorrelatedTestEvent>
{
    public async Task Consume(ConsumeContext<CorrelatedTestEvent> context)
    {
        // 直接从当前上下文获取初始CorrelationId
        var correlationId = context.CorrelationId.Value;

        // 使用context.Publish发布后续事件，自动传递关联信息
        await context.Publish(new CorrelatedTestEvent2 
        { 
            CorrelationId = correlationId 
        }, context.CancellationToken);
    }
}

public class CorrelatedTestEvent2Consumer : IConsumer<CorrelatedTestEvent2>
{
    public async Task Consume(ConsumeContext<CorrelatedTestEvent2> context)
    {
        // 这里可以获取到和初始事件相同的CorrelationId
        var correlationId = context.CorrelationId.Value;

        await context.Publish(new CorrelatedTestEvent3 
        { 
            CorrelationId = correlationId 
        }, context.CancellationToken);
    }
}

3. 若必须使用IMediator.Publish，手动传递CorrelationId #

如果因业务需求必须直接使用IMediator发布，需要在发布时手动设置新事件的CorrelationId上下文：

public class CorrelatedTestEventConsumer : IConsumer<CorrelatedTestEvent>
{
    private readonly IMediator _mediator;

    public CorrelatedTestEventConsumer(IMediator mediator)
    {
        _mediator = mediator;
    }

    public async Task Consume(ConsumeContext<CorrelatedTestEvent> context)
    {
        var correlationId = context.CorrelationId.Value;

        await _mediator.Publish(new CorrelatedTestEvent2 
        { 
            CorrelationId = correlationId 
        }, 
        publishContext => publishContext.CorrelationId = correlationId, 
        context.CancellationToken);
    }
}

可行性说明 #

完全可行，MassTransit Mediator的设计目标之一就是支持进程内事件的关联追踪，只要遵循上述上下文传递的方式，就能保证同一操作链中所有事件的ConsumeContext都能拿到初始的CorrelationId，进而实现整体进度追踪和操作完成判断。

内容的提问来源于stack exchange，提问作者Dovic