.NET Core中RabbitMQ消息驱动调用CQRS方法的最佳实践

我在.NET Core的CQRS架构Presentation层实现了RabbitMQ队列的消息监听器，代码如下：

//CQRS架构的Presentation层
public async Task ReceiveAsync<T>(string queue, Action<T> onMessage)
{
    _channel.QueueDeclare(queue, true, false, false);
    var consumer = new AsyncEventingBasicConsumer(_channel);
    consumer.Received += async (s, e) =>
    {
        var jsonSpecified = Encoding.UTF8.GetString(e.Body.Span);
  
        var item = JsonSerializer.Deserialize<RabbitRecieverWrapperModel>(jsonSpecified);
        
        string methodName = "GetTypes";

        //使用MethodInfo类获取方法信息
        MethodInfo mi = this.GetType().GetMethod(methodName);

        //调用方法
        // (null表示无参数，也可传入参数数组...)
        mi.Invoke(this, null);
        //onMessage(item);

        await Task.Yield();
    };
    _channel.BasicConsume(queue, true, consumer);
    await Task.Yield();
}

需要根据RabbitMQ收到的消息调用的方法位于Core项目中，例如以下CQRS Query及其Handler实现：

namespace SefService.Application.Invoices.Queries
{
    public class GetUnitOfMeasuresQuery : IRequest<Result<IList<UnitOfMeasureResponseDto>>>
    {
        
    }

    public class GetUnitOfMeasuresQueryHandler : IRequestHandler<GetUnitOfMeasuresQuery, Result<IList<UnitOfMeasureResponseDto>>>
    {
        public ILogger<UnitOfMeasureResponseDto> _logger { get; }
        private readonly ISefProxy _sefProxy;

        public GetUnitOfMeasuresQueryHandler(ISefProxy sefProxy, ILogger<UnitOfMeasureResponseDto> logger)
        {
            _logger = logger;
            _sefProxy = sefProxy;
        }

        public async Task<Result<IList<UnitOfMeasureResponseDto>>> Handle(GetUnitOfMeasuresQuery request, CancellationToken cancellationToken)
        {
            try
            {
                if (request is null)
                {
                    throw new ArgumentNullException(nameof(request));
                }

                var response = await _sefProxy.GetUnitOfMeasures();
                return new Result<IList<UnitOfMeasureResponseDto>> { Data = response.Data };
            }
            catch (System.Exception ex)
            {
                _logger.LogError(ex.Message, ex);
                throw new ApiException(ex.Message);
            }
        }
    }
}

该方法在标准WebApi（Swagger）中的调用方式如下：

/// <summary>
/// GetUnitOfMeasures
/// </summary>
/// <returns></returns>
[ProducesResponseType(StatusCodes.Status200OK)]
[ProducesResponseType(StatusCodes.Status400BadRequest)]
[ProducesResponseType(StatusCodes.Status500InternalServerError)]
[ProducesDefaultResponseType]
[Route(nameof(GetUnitOfMeasures))]
[HttpGet]
public async Task<IActionResult> GetUnitOfMeasures()
{
    return Ok(await Mediator.Send(new GetUnitOfMeasuresQuery()));
}

我已尝试使用反射实现，但需要跨项目对CQRS模式下的Command类使用反射。请问我应继续采用反射，还是使用责任链模式？此场景下的最佳实践是什么？

1. 优先复用MediatR的调度能力（最佳实践） #

你的项目已经基于MediatR实现CQRS，这是核心调度器，没必要重新造轮子。RabbitMQ消息监听的核心应该是把消息反序列化为对应的Command/Query对象，直接交给MediatR处理，完全贴合现有架构：

• 步骤1：在RabbitRecieverWrapperModel中增加字段存储Command/Query的完全限定类型名（如SefService.Application.Invoices.Queries.GetUnitOfMeasuresQuery）
• 步骤2：在消息消费逻辑中，加载对应类型并通过MediatR调度：

// 从配置或依赖注入获取Core项目程序集
var coreAssembly = Assembly.Load("SefService.Application");
// 根据类型名加载具体Command/Query类型
var commandType = coreAssembly.GetType(item.CommandTypeFullName);
// 反序列化为具体实例
var command = JsonSerializer.Deserialize(jsonSpecified, commandType);
// 注入IMediator后直接发送请求
await _mediator.Send(command, cancellationToken);

• 优势：复用现有Handler的依赖注入、异常处理、管道行为，架构一致性强，无需额外维护映射关系。

2. 反射方案的局限性 #

直接反射调用Handler的Handle方法会跳过MediatR的生命周期管理，导致代码重复且不易维护。即使优化反射逻辑，最终还是绕回MediatR的调度，不如直接用MediatR的动态发送简洁。

3. 责任链模式的适用场景 #

责任链适合消息的多步骤处理（如验证、日志、格式转换），但对于“根据消息类型匹配CQRS Handler”的核心需求，会需要手动维护每个Command/Query对应的处理节点，随着业务扩展，维护成本会急剧上升，远不如MediatR的自动注册高效。

4. 额外优化建议 #

• 避免硬编码程序集名称，可通过配置文件或依赖注入获取程序集实例
• 增加消息类型校验逻辑，防止无效消息引发异常
• 用Polly等库实现重试、熔断机制，提升消息消费可靠性

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