Azure Service Bus与MassTransit的iConsumer消费后业务逻辑执行问题

你所说的worker类（即MassTransit中实现了IConsumer<T>接口的消费者类）就是消息消费完成后启动域内业务逻辑的标准入口。

核心逻辑说明 #

• MassTransit 已经内置了Azure Service Bus的消息适配逻辑，传输的JSON payload会被自动反序列化为你定义的强类型消息对象，你可以直接从ConsumeContext<T>的Message属性中拿到完整的消息内容，不需要手动处理JSON解析、基础消息确认等操作。
• 不建议把核心业务逻辑直接写在Consume方法中，推荐通过依赖注入把对应的领域服务、业务处理类注入到消费者中，Consume方法仅作为入口做参数校验、上下文传递，再调用对应的业务方法即可，这样可以保证业务逻辑和消息队列组件解耦，后续调整消息队列方案时不需要修改核心业务代码。
• 你也可以根据业务需要在消费者中补充额外的处理逻辑：比如消费前的幂等性校验、异常重试失败后的死信队列转发、手动控制消息确认时机等。

简单实现示例 #

// 消费者类（即示例中常见的worker类）
public class OrderCreatedConsumer : IConsumer<OrderCreated>
{
    // 注入域内业务服务
    private readonly IOrderProcessService _orderProcessService;

    public OrderCreatedConsumer(IOrderProcessService orderProcessService)
    {
        _orderProcessService = orderProcessService;
    }

    public async Task Consume(ConsumeContext<OrderCreated> context)
    {
        // 获取已经反序列化完成的强类型消息
        var orderMsg = context.Message;
        // 启动域内业务逻辑
        await _orderProcessService.HandleNewOrder(orderMsg.OrderId, orderMsg.Amount);
    }
}

如果你的项目采用DDD架构，也可以在消费者中仅做消息转换，把消息队列的外部事件转为域内的本地事件之后发布，由域内的事件处理器来执行业务逻辑，进一步拆分职责。

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