Maomi.MQ 功能强大的 .NET RabbitMQ 消息队列通讯模型框架完整指南-45看点

平时做技术实践时，很多问题不是概念不会，而是细节没串起来。拿“Maomi.MQ 功能强大的 .NET RabbitMQ 消息队列通讯模……”来说，它看着像小点，放到项目里常会牵出环境、配置、兼容性和维护成本。下面按实际采用顺序，把思路、关键写法和容易踩坑的地方讲清楚，便于大家直接对照操作。

更快开始

在这个场景下，在本篇教程里，将介绍 Maomi.MQ.RabbitMQ 的采用方法，以便读者能够更快了解该框架的采用方式和特点。

从实现思路看，Maomi.MQ.RabbitMQ 是一个基于 RabbitMQ 的消息队列封装框架，提供了很多开箱即用的功能，借助轻松灵活的方式简化消息传输流程，提供一系列可靠的消息传输保障机制，降低开发者采用难度，减少开发时间。

主要功能：

简化的消息定义和消费者，无需复杂设置即可上手。
结合项目来看，丰富灵活的设置，发挥 RabbitMQ 的强大力量，自动新建队列、死信队列、广播模式、Qos 同时发控制、动态路由、动态消费者。
在这个场景下，自定义序列化器，兼容 Json、Protobuf、Thrift、MessagePack 二进制消息传输，高性能压缩消息减少内存、提高同时发量、跨微服务传输。
实际处理时，兼容自动和自定义发布消息推送交换器，兼容 RabbitMQ 事务推送模式。
结合项目来看，简化的消费者模式、事件总线模式、广播模式、动态消费者，还兼容结合 MeditaR、FastEndpoints 框架采用，进一步减少采用负担，减少代码侵入性。
自定义重试策略，从容应对服务错误、强一致性消息、高同时发流量。
实际处理时，兼容本地消息表模式，强一致性保证业务消息不丢失，解决空悬挂、重复处理等异步消息异常问题，保证业务可靠性。
结合项目来看，自由的消费者模式，除了 MeditaR、FastEndpoints，还能够自由接入其它框架，充分借助第三方框架的优质能力。

更快设置

理解这一步时，新建一个 Web 项目（可参考 WebDemo 项目），引入 Maomi.MQ.RabbitMQ 包，在 Web 设置中注入服务：

// using Maomi.MQ;
// using RabbitMQ.Client;
builder.Services.AddMaomiMQ((MqOptionsBuilder options) =>
{
    options.WorkId = 1;
    options.AppName = "myapp";
    options.Rabbit = (ConnectionFactory options) =>
    {
        options.HostName = Environment.GetEnvironmentVariable("RABBITMQ")!;
        options.Port = 5672;
    };
}, [typeof(Program).Assembly]);
var app = builder.Build();

WorkId：指定用来生成分布式雪花 id 的节点 id，默认为 0。
在这个场景下，每条消息生成一个唯一的 id，便于追踪。如果不设置雪花id，在分布式服务中，多实例同时行工作时，可能会产生相同的 id。
AppName：用来标识在日志和链路追踪中标识消息的生产者或消费者。
Rabbit：RabbitMQ 客户端设置，请参考 ConnectionFactory。

实际处理时，定义消息模型类，模型类是 MQ 通讯的消息基础，该模型类将会被序列化为二进制内容传递到 RabbitMQ 服务器中。

public class TestEvent
{
    public int Id { get; set; }
    public override string ToString()
    {
        return Id.ToString();
    }
}

定义消费者，消费者需实现 IConsumer 接口，以及采用 [Consumer] 特性注解设置消费者属性，如下所示所示，[Consumer("test")] 表示该消费者订阅的队列名称是 test。

IConsumer 接口有三个方法，ExecuteAsync 方法用来处理消息，FaildAsync 会在 ExecuteAsync 异常时立即执行，如果代码一直异常，最后会调用 FallbackAsync 方法，Maomi.MQ 框架会根据 ConsumerState 值确定是否将消息放回队列重新消费，或者做其它处理动作。

[Consumer("test")]
public class MyConsumer : IConsumer
{
    // 消费
    public async Task ExecuteAsync(MessageHeader messageHeader, TestEvent message)
    {
        Console.WriteLine($"事件 id: {message.Id} {DateTime.Now}");
        await Task.CompletedTask;
    }
    // 每次消费失败时执行
    public Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, TestEvent message) 
        => Task.CompletedTask;
    // 补偿
    public Task FallbackAsync(MessageHeader messageHeader, TestEvent? message, Exception? ex) 
        => Task.FromResult( ConsumerState.Ack);
}

落到代码里，Maomi.MQ 还具有多种消费者模式，代码写法不一样，后续会细致说明不同的消费者模式。

理解这一步时，若要发布消息，只需注入 IMessagePublisher 服务即可。

private readonly IMessagePublisher _messagePublisher;
public IndexController(IMessagePublisher messagePublisher)
{
    _messagePublisher = messagePublisher;
}
[HttpGet("publish")]
public async Task Publisher()
{
    // 发布消息
    await _messagePublisher.PublishAsync(exchange: string.Empty, routingKey: "test", message: new TestEvent
    {
        Id = 123
    });
    return "ok";
}

在这个场景下，启动 Web 服务，在 swagger 页面上请求 API 接口，MyConsumer 服务会立即接收到发布的消息。

就是这么轻松，就是这么便于。

怎么发布消息

自动发布

落到代码里，虽然发布者和消费者共用一个模型类，但是在一个项目中怎么设置模型类，都不会影响消费者。将分布者消费者隔离简化框架设计和微服务解耦，兼容不同的编程语言服务相互通讯，共同完成业务逻辑。

比如为了简化消息发布，我们能够在模型类指定绑定的路由键。

[RouterKey("scenario.quickstart")]
public sealed class QuickStartMessage
{
}

发布时是根据模型类的 [RouterKey] 自动找到要推送的交换器和路由键，简化发布消息的参数。

await _publisher.AutoPublishAsync(message);

当然手动设置也能够：

await _publisher.PublishAsync(string.Empty, request.Queue, message);

实际处理时，消费者能够自由设置要消费的队列，即使是相同的模型类，能够自由设定队列名称，不会产生干扰。

[Consumer("scenario.quickstart")]
public sealed class QuickStartConsumer : IConsumer
{
}

手动发布

适用来你需精细控制 exchange/routingKey 或消息属性（TTL、Header、优先级等）的场景。

[HttpPost("publish-manual")]
public async Task PublishManual()
{
    var message = new OrderCreatedMessage
    {
        OrderNo = "SO-20260211-002",
        Amount = 299.00m
    };
    await _publisher.PublishAsync(
        exchange: "biz.order.exchange",
        routingKey: "order.created.v1",
        message: message,
        properties: p =>
        {
            p.Expiration = "60000";
            p.Headers ??= new Dictionary();
            p.Headers["tenant"] = "tenant-a";
        });
    return Results.Ok(message);
}

理解这一步时，但是 Maomi.MQ.RabbitMQ 提供了更为轻松易用的方式，实现自动处理队列属性。

比如，要设计一个死信队列，你只需在消费者上设置属性即可：

[Consumer(
    "example.retry.main",
    RetryFaildRequeue = false,
    DeadExchange = "",
    DeadRoutingKey = "example.retry.dead")]
public sealed class RetryConsumer : IConsumer
{
}

RetryFaildRequeue 表示消费失败后不会放回原队列。
DeadExchange 死信交换器名称。
DeadRoutingKey 死信队列路由键。

所以，发布消息时，只需采用这两行代码：

await _publisher.PublishAsync(string.Empty, request.Queue, message);
await _publisher.AutoPublishAsync(message);

普通消费者、动态消费者、事件总线

实际处理时，Maomi.MQ 兼容以多种姿势新建消费者，一个项目里面能够同时采用多种消费者模式，自由灵活而不会冲突。

这三种模式不是互斥关系，而是处理问题的方式不同：

普通消费者：最基础的消费模式，继承 IConsumer 即可采用，还能够自由扩展出不同类型的消费者模式，比如 MediatR。
动态消费者：也是继承 IConsumer，运行时动态新建/停止订阅。
事件总线：一个消息触发多个有顺序的处理步骤，生成执行链和回滚链路。

普通消费者模式

实现 IConsumer 接口即可，能够自定义消费、重试、回滚逻辑，这种模式采用轻松，还能从容处理消息错误和回滚。

正常处理消息会调用 ExecuteAsync，失败后调用 FaildAsync，重试耗尽进入 FallbackAsync。

[Consumer("biz.order.created.v1", Qos = 10, RetryFaildRequeue = false)]
public sealed class NormalOrderConsumer : IConsumer
{
    public Task ExecuteAsync(MessageHeader messageHeader, OrderCreatedMessage message)
    {
        Console.WriteLine($"normal consume => {message.OrderNo}");
        return Task.CompletedTask;
    }
    public Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, OrderCreatedMessage message)
        => Task.CompletedTask;
    public Task FallbackAsync(MessageHeader messageHeader, OrderCreatedMessage? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}

动态消费者模式

落到代码里，你能够在运行时将业务需订阅的消息队列动态注册，比如 SAAS 平台新建租户后需动态按照租户前缀消费对应的主题，也能够自行取消订阅。

public sealed class DynamicDemoService
{
    private readonly IDynamicConsumer _dynamicConsumer;
    public DynamicDemoService(IDynamicConsumer dynamicConsumer)
    {
        _dynamicConsumer = dynamicConsumer;
    }
    public async Task StartAsync(string queue)
    {
        var options = new ConsumerAttribute(queue) { Qos = 5 };
        var consumerTag = await _dynamicConsumer.ConsumerAsync(
            options,
            execute: (header, message) =>
            {
                Console.WriteLine($"dynamic consume => {message.OrderNo}");
                return Task.CompletedTask;
            },
            faild: (header, ex, retryCount, message) => Task.CompletedTask,
            fallback: (header, message, ex) => Task.FromResult(ConsumerState.Ack));
        return consumerTag;
    }
    public Task StopByQueueAsync(string queue)
        => _dynamicConsumer.StopConsumerAsync(queue);
}

事件总线模式

理解这一步时，事件总线模式能够自由编排事件执行链路，框架会按照链路自动执行同时在执行失败后自动执行回滚链路。

IEventMiddleware 负责构建执行链，[EventOrder] 控制步骤顺序，适合一个事件拆成多个业务步骤，能够很好将业务解耦。

using Maomi.MQ.EventBus;
[RouterKey("biz.order.pipeline.v1")]
public sealed class OrderPipelineEvent
{
    public Guid OrderId { get; set; } = Guid.NewGuid();
    public decimal Amount { get; set; }
}
[Consumer("biz.order.pipeline.v1")]
public sealed class OrderPipelineMiddleware : IEventMiddleware
{
    public Task ExecuteAsync(MessageHeader messageHeader, OrderPipelineEvent message, EventHandlerDelegate next)
    {
        return next(messageHeader, message, CancellationToken.None);
    }
    public Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, OrderPipelineEvent? message)
        => Task.CompletedTask;
    public Task FallbackAsync(MessageHeader messageHeader, OrderPipelineEvent? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}
[EventOrder(1)]
public sealed class ReserveInventoryHandler : IEventHandler
{
    public Task ExecuteAsync(OrderPipelineEvent message, CancellationToken cancellationToken)
        => Task.CompletedTask;
    public Task CancelAsync(OrderPipelineEvent message, CancellationToken cancellationToken)
        => Task.CompletedTask;
}
[EventOrder(2)]
public sealed class CreateBillHandler : IEventHandler
{
    public Task ExecuteAsync(OrderPipelineEvent message, CancellationToken cancellationToken)
        => Task.CompletedTask;
    public Task CancelAsync(OrderPipelineEvent message, CancellationToken cancellationToken)
        => Task.CompletedTask;
}

广播模式

理解这一步时，比如为了减少延时和提供性能，在 Redis 做完缓存后，还需做本地缓存。但是如果数据发生变更，怎么刷新本地缓存呢？

实际处理时，那就采用广播模式，同一个服务的不同实例采用广播模式时，每个实例都能够收到消息，而不是随机分配给其中一个。

代码很轻松，设置 IsBroadcast = true 即可，这样即使是同一个服务的不同实例，也会收到广播通知，该实例下线，会自动取消订阅，不会消耗服务器资源。

[Consumer("scenario.quickstart", IsBroadcast = true)]
public sealed class QuickStartConsumer : IConsumer
{
}

高性能序列化器

落到代码里，Maomi.MQ 默认采用 JSON 做序列化数据传输，你也能够引入其它序列化器，提高压缩消息的性能。

理解这一步时，目前兼容 System.Text.Json、Protobuf、Thrift、MessagePack 四种二进制序列化协议，你能够自由选择组合采用不同的序列化器到项目中，以便在不同微服务中传递消息，同时且实现高性能传递消息。

落到代码里，比如采用 protobuf-net 框架识别标记了 [ProtoContract] 的模型类，那么此类型采用 Protobuf 协议压缩消息，其它消息还是走 JSON。

using ProtoBuf;
builder.Services.AddMaomiMQ(options =>
{
    options.MessageSerializers = serializers =>
    {
        // 添加 Protobuf 序列化器
        serializers.Insert(0, new ProtobufMessageSerializer());
    };
}, [typeof(Program).Assembly]);
[ProtoContract]
public sealed class PersonMessage
{
    [ProtoMember(1)]
    public Guid Id { get; set; } = Guid.NewGuid();
    [ProtoMember(2)]
    public string Name { get; set; } = string.Empty;
    [ProtoMember(3)]
    public int Age { get; set; }
}

强一致性事务模式

理解这一步时，借鉴 CAP 等框架的本地消息表模式，借助 MQ 和本地消息表，实现轻松的强一致性的分布式事务，在业务不太复杂的企业项目中，能够简化编写事务的难度，不同的微服务以轻量、简洁、不复杂的模式接入，降低了编码和维护难度。

设置（以 MySQL 为例）：

using Maomi.MQ.Transaction.Mysql;
using MySqlConnector;
builder.Services.AddMaomiMQTransactionMySql();
builder.Services.AddMaomiMQTransaction(options =>
{
    options.ProviderName = TransactionProviderNames.MySql;
    options.Connection = _ => new MySqlConnection(builder.Configuration.GetConnectionString("Default"));
    options.AutoCreateTable = true;
});

业务代码发布消息：

public sealed class OrderAppService
{
    private readonly IMessagePublisher _publisher;
    private readonly string _connectionString;
    public OrderAppService(IMessagePublisher publisher, IConfiguration configuration)
    {
        _publisher = publisher;
        _connectionString = configuration.GetConnectionString("Default")!;
    }
    public async Task CreateOrderAsync(CancellationToken cancellationToken)
    {
        await using var connection = new MySqlConnection(_connectionString);
        await connection.OpenAsync(cancellationToken);
        await using var transaction = await connection.BeginTransactionAsync(cancellationToken);
        // 1) 执行业务 SQL
        // await SaveOrderAsync(connection, transaction, ...);
        await transaction.CommitAsync(cancellationToken);      
        // 2) 发送消息（放在事务外面）
        await _publisher.AutoPublishAsync(new OrderCreatedMessage
        {
            OrderNo = "SO-TX-001",
            Amount = 520m
        }, cancellationToken: cancellationToken);
    }
}

注意：本地事务模式不是全局分布式事务协调器，它解决的是“单库与消息发送”的一致性。

在这个场景下，若事务提交了但是最后发送消息失败，那么后台会有一个 BackgroundService 定期扫描数据库，将这些没有发送成功的消息推送到 RabbitMQ。

结合 MediatR

落到代码里，若你的业务已经采用 MediatR，能够直接把 MQ 当作 MediatR 命令输入通道，也就是原有的代码完全不需改动，只需在 Command 上加上 [MediatRConsumer] 即可。

using Maomi.MQ.MediatR;
using MediatR;
[MediatRConsumer("biz.mediatr.order", Qos = 1)]
public sealed class SyncOrderCommand : IRequest
{
    public string OrderNo { get; set; } = string.Empty;
}
public sealed class SyncOrderCommandHandler : IRequestHandler
{
    public Task Handle(SyncOrderCommand request, CancellationToken cancellationToken)
        => Task.CompletedTask;
}
// 通过 MediatR 触发 MQ 发布
await mediator.Send(new MediatRMqCommand
{
    Message = new SyncOrderCommand { OrderNo = "SO-MED-001" }
});

原理：MediatRTypeFilter 会把带 MediatRConsumer 的命令类型映射成 MQ 消费者，消费后再转发给 IMediator.Send(...)。

或者你能够继续采用 IMessagePublisher 发布消息。

await _publisher.PublishAsync(string.Empty, request.Queue, message);

落到代码里，若你的项目已经引入了 MediatR，那么不需为了采用 RabbitMQ 再搞出别的消费模式代码，采用 Maomi.MQ 能够直接简化接入 RabbitMQ 的麻烦，继续以 MediatR 的模式实现异步消费。

结合 FastEndpoints

落到代码里，若你采用 FastEndpoints，也能够借助类型过滤器把 IEvent/ICommand 接入 MQ。

using Maomi.MQ.Filters;
app.Services.RegisterGenericCommand(typeof(FastEndpointsMqCommand<>), typeof(FastEndpointsMqCommandHandler<>));
app.UseFastEndpoints();
[FastEndpointsConsumer("biz.fast.order.paid", Qos = 1)]
public sealed class OrderPaidEvent : FastEndpoints.IEvent
{
    public string OrderNo { get; set; } = string.Empty;
}

await _messagePublisher.AutoPublishAsync(new OrderPaidEvent
{
    OrderNo = "SO-FE-001"
});

其它能力

下面这些能力这里先简述，后续你能够按需深入：

动态路由设置：实现 IRoutingProvider，统一改写 Exchange/RoutingKey/Queue。
自定义重试机制：实现 IRetryPolicyFactory，按队列/类型定制重试策略。
宽松自由可定制消费者模式：实现 ITypeFilter，把第三方框架类型映射成 Maomi.MQ 消费者。

到此这篇关于Maomi.MQ 功能强大的 .NET RabbitMQ 消息队列通讯模型框架来了的文章就介绍到这了,更多相关Maomi.MQ 功能强大的 .NET RabbitMQ 消息队列通讯模型框架来了内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多兼容脚本之家！

您可能感兴趣的文章:

.NET Core实现RabbitMQ消息队列的示例
RabbitMQ .NET消息队列采用详解
Maomi.MQ 功能强大的 .NET RabbitMQ 消息队列通讯模型框架详解
asp.net借助消息队列处理高同时发请求(以抢小米手机为例)
.NetCore借助BlockingCollection实现简易消息队列
.net msmq消息队列实例详解
.Net消息队列的采用方法

Maomi.MQ 功能强大的 .NET RabbitMQ 消息队列通讯模型框架完整指南

目录