.NET+MassTransit：控制器动作如何等待消息驱动工作流结果？

一、让AddArticle动作等待结果消息的实现方式 #

在MassTransit中，你可以通过关联ID（CorrelationId）+ 请求/响应地址绑定的方式，让控制器动作跟踪并等待工作流的最终结果。核心思路是：在初始消息中携带唯一关联ID和响应地址，让工作流的最终节点（CalcB）将结果发回指定地址，控制器通过该地址等待结果。

具体实现代码 #

1. 控制器动作修改

[HttpPost]
public async Task<ActionResult<ResultMessage>> PostAsync([FromBody] ArticleCreateRequest request, CancellationToken cancellationToken)
{
    // 生成唯一关联ID，用于跟踪整个工作流
    var correlationId = Guid.NewGuid();

    // 创建请求客户端，指定等待ResultMessage，同时设置超时时间
    var requestClient = _bus.CreateRequestClient<ResultMessage>(TimeSpan.FromSeconds(30), cancellationToken);

    // 向CalcA发布初始处理命令，携带关联ID和响应地址
    await _bus.Publish(new ProcessArticleInitiateCommand
    {
        CorrelationId = correlationId,
        ArticleContent = request.Content,
        // 将请求客户端的地址传递给工作流节点，用于后续返回结果
        ResponseAddress = requestClient.Address
    }, cancellationToken);

    // 等待带有指定关联ID的结果消息
    var response = await requestClient.GetResponse<ResultMessage>(
        filter => filter.CorrelationId == correlationId, 
        cancellationToken
    );

    return Ok(response.Message);
}

2. CalcB消费者处理结果返回

public class CalcBConsumer : IConsumer<CalcBProcessingCommand>
{
    public async Task Consume(ConsumeContext<CalcBProcessingCommand> context)
    {
        // 执行业务逻辑，生成结果
        var result = new ResultMessage
        {
            CorrelationId = context.Message.CorrelationId,
            Success = true,
            ArticleId = Guid.NewGuid(),
            Message = "文章处理完成"
        };

        // 将结果发送到初始请求指定的响应地址
        await context.Send(context.Message.ResponseAddress, result);
    }
}

注意：需要确保ProcessArticleInitiateCommand和CalcBProcessingCommand都携带CorrelationId和ResponseAddress字段，在CalcA向CalcB转发消息时，要完整传递这两个参数。

二、请求/响应模式是否适合大型工作流？ #

这取决于你对“大型工作流”的定义，分两种场景讨论：

1. 适合的场景 #

如果工作流满足以下条件，请求/响应模式完全可用：

• 总耗时在HTTP请求超时范围内（通常30秒以内）
• 步骤相对简单，节点间依赖关系清晰
• 对实时性要求高，客户端需要立即获取结果

MassTransit的请求/响应模式内置了超时控制、重试机制，能保证消息的可靠性传递，足以应对这类场景。

2. 不适合的场景（推荐替代方案） #

如果工作流属于长耗时、多节点、高复杂度类型（比如耗时超过1分钟，涉及5个以上服务节点，需要处理失败补偿），请求/响应模式会暴露以下问题：

• HTTP连接超时风险：客户端或服务器端的超时设置会导致请求失败，而工作流可能仍在后台执行
• 资源占用过高：WebApi线程被长时间占用，降低系统并发能力
• 状态一致性难保障：中间节点故障时，HTTP请求直接失败，但工作流状态无法回滚或跟踪

这种情况下，推荐采用异步请求+状态查询的方案，结合MassTransit的Saga（传奇模式）管理工作流：

1. 控制器接收请求后，立即返回一个唯一的任务ID给客户端
2. 启动MassTransit Saga，协调CalcA、CalcB等节点的工作流执行，实时更新工作流状态到数据库
3. 客户端通过任务ID轮询控制器获取结果，或通过WebSocket/WebHook接收完成通知
4. Saga负责处理工作流中的失败重试、补偿逻辑，保证流程的最终一致性

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