31. RabbitMQ 消息队列
后端手册RabbitMQ消息队列大约 8 分钟约 2301 字
一句话定位：RabbitMQ 在 Admin.NET 里承担两个互不相干的职责——(a) 作为事件总线的事件源存储器（跨进程可靠队列），(b) 作为业务消息队列（RabbitMqProducer 生产 + IMessageHandler 消费）。本篇讲后者。
一、职责边界（别搞混）
┌───────────────────────────────────────────────────────────────┐
│ ① 事件总线（进程内 / Redis / RabbitMQ 作为"存储器"） │
│ IEventPublisher.PublishAsync(eventId, payload) │
│ → 同进程内的 [EventSubscribe] 处理器消费 │
│ 用途：日志落库、发邮件、发短信（详见第 18 篇） │
│ EventBus:EventSourceType = Memory | Redis | RabbitMQ │
└───────────────────────────────────────────────────────────────┘
┌───────────────────────────────────────────────────────────────┐
│ ② 业务消息队列（本篇） │
│ RabbitMqProducer.SendMessage(queueName, message) │
│ → IMessageHandler.HandleMessageAsync(message) 消费 │
│ 用途：跨服务解耦、削峰、异步任务派发 │
│ 开关：同一个 EventBus:EventSourceType == "RabbitMQ" │
└───────────────────────────────────────────────────────────────┘两者的耦合点只有一个：RabbitMQSetup.AddRabbitMQ() 内部会先检查 EventBus:EventSourceType != "RabbitMQ" 则直接 return，所以没开启 RabbitMQ 事件源时，业务队列消费者也一个都不会启。
// Admin.NET.Core/RabbitMQ/RabbitMQSetup.cs
public static void AddRabbitMQ(this IServiceCollection services, int consumerCount = 1)
{
if (App.GetConfig<string>("EventBus:EventSourceType", true) != "RabbitMQ") return;
// 注册多个消费者服务
for (int i = 0; i < consumerCount; i++)
services.AddHostedService<RabbitMqConsumer>();
}二、配置
// Admin.NET.Application/Configuration/MessageQueue.json
{
"$schema": ".../furion-schema.json",
// 事件源存储器配置
"EventBus": {
"EventSourceType": "Memory" // Memory、Redis、RabbitMQ、Kafka
},
// RabbitMQ 连接配置
"RabbitMQ": {
"UserName": "adminnet",
"Password": "adminnet++123456",
"HostName": "127.0.0.1",
"Port": 5672,
"VirtualHost": "/"
},
// RabbitMQ 消费者配置
"RabbitMqConsumerOptions": {
"ConsumerCount": 1 // 消费者数量，多消费者实现多线程消费队列
}
}绑定类：
// Admin.NET.Core/RabbitMQ/RabbitMqOptions.cs
public class RabbitMqOptions : IConfigurableOptions
{
public string UserName { get; set; }
public string Password { get; set; }
public string HostName { get; set; }
public int Port { get; set; }
public string VirtualHost { get; set; } = "/";
}
// Admin.NET.Core/RabbitMQ/RabbitMqConsumerOptions.cs
public class RabbitMqConsumerOptions : IConfigurableOptions
{
public string ConsumerCount { get; set; } // 注意：字符串类型
}配置注册在 Admin.NET.Web.Core/ProjectOptions.cs：
services.AddConfigurableOptions<RabbitMqOptions>();
services.AddConfigurableOptions<RabbitMqConsumerOptions>();注册与读取消费者数量（Admin.NET.Application/Startup.cs）：
// 注册 RabbitMQ 服务
var option = App.GetConfig<RabbitMqConsumerOptions>("RabbitMqConsumerOptions");
var consumerCount = option.ConsumerCount.ToIntOrDefault(1);
services.AddRabbitMQ(consumerCount);注意配置节点名是
RabbitMqConsumerOptions（驼峰中间小写 q），而类名是RabbitMqConsumerOptions，两者一致；写错节点名会得到默认值 1。
三、生产者 RabbitMqProducer
// Admin.NET.Core/RabbitMQ/RabbitMqProducer.cs
public class RabbitMqProducer : ISingleton
{
private readonly RabbitMqConnection _connection;
public async Task SendMessage(string queueName, string message)
{
var channel = _connection.Channel;
await channel.QueueDeclareAsync(queue: queueName, durable: true, exclusive: false,
autoDelete: false, arguments: null);
var properties = new BasicProperties { Persistent = true }; // 持久化
var body = Encoding.UTF8.GetBytes(message);
await channel.BasicPublishAsync(exchange: string.Empty, routingKey: queueName,
mandatory: true, basicProperties: properties, body: body);
}
}特征：默认交换机（"", direct）+ routingKey = queueName，durable: true + Persistent = true 双保险持久化，mandatory: true 表示无队列时会被 Broker 退回。
调用示例：
[ApiDescriptionSettings(Order = 910, Description = "订单")]
public class BizOrderService : IDynamicApiController, ITransient
{
private readonly RabbitMqProducer _producer;
public BizOrderService(RabbitMqProducer producer) => _producer = producer;
/// <summary>
/// 提交订单 🔖（异步出库仓储，削峰）
/// </summary>
[DisplayName("提交订单")]
public async Task Submit(SubmitOrderInput input)
{
// 1. 同步落库
var order = input.Adapt<BizOrder>();
await _rep.InsertAsync(order);
// 2. 投递消息给下游 WMS
await _producer.SendMessage("order_created", JSON.Serialize(new
{
order.Id, order.OrderNo, order.Amount, input.WarehouseId
}));
}
}四、消费者：实现 IMessageHandler
// Admin.NET.Core/RabbitMQ/IMessageHandler.cs
public interface IMessageHandler : ITransient
{
string QueueName { get; }
Task HandleMessageAsync(string message);
}实现示例：
using Newtonsoft.Json.Linq;
namespace Admin.NET.Application;
/// <summary>
/// 订单创建消息处理器
/// </summary>
public class OrderCreatedHandler : IMessageHandler
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly ILogger _logger;
public OrderCreatedHandler(IServiceScopeFactory scopeFactory, ILoggerFactory loggerFactory)
{
_scopeFactory = scopeFactory;
_logger = loggerFactory.CreateLogger(CommonConst.SysLogCategoryName); // 会写入数据库日志
}
public string QueueName => "order_created";
public async Task HandleMessageAsync(string message)
{
try
{
var jo = JObject.Parse(message);
var orderId = jo["Id"]!.Value<long>();
// 后台线程必须用 Scope + CopyNew()，见第 18 篇「后台线程使用仓储的坑」
using var scope = _scopeFactory.CreateScope();
var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();
await db.Updateable<BizOrder>()
.SetColumns(u => u.Status == OrderStatusEnum.Pushed)
.Where(u => u.Id == orderId)
.ExecuteCommandAsync();
}
catch (Exception ex)
{
// 抛异常 → 框架会 BasicNack(requeue: true) 重新入队
_logger.LogError(ex, "处理订单创建消息失败：{Message}", message);
throw;
}
}
}无需手写注册——IMessageHandler : ITransient 实现了 Furion 的瞬态生命周期标记，启动时随程序集自动扫描注入，RabbitMqConsumer 构造时从容器拿到 最后一个注册 的实现。
五、RabbitMqConsumer 源码要点
// Admin.NET.Core/RabbitMQ/RabbitMqConsumer.cs
public class RabbitMqConsumer : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
if (!await _connection.TryConnectAsync())
{
_logger.LogError("RabbitMQ连接失败，请检查配置文件。");
return; // 连不上就退出，不崩溃
}
var channel = _connection.Channel;
await channel.QueueDeclareAsync(queue: _handler.QueueName, durable: true, exclusive: false,
autoDelete: false, arguments: null);
await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false); // 每次只拿 1 条
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (sender, ea) =>
{
try
{
var message = Encoding.UTF8.GetString(ea.Body.ToArray());
await _handler.HandleMessageAsync(message);
Console.WriteLine($"[x] Received from {_handler.QueueName}: {message}");
await channel.BasicAckAsync(deliveryTag: ea.DeliveryTag, multiple: false); // 手动 ACK
}
catch (Exception ex)
{
_logger.LogError(ex, "消息处理失败");
await channel.BasicNackAsync(deliveryTag: ea.DeliveryTag, multiple: false, requeue: true); // 重新入队
}
};
await channel.BasicConsumeAsync(queue: _handler.QueueName, autoAck: false, consumer: consumer);
// ...stoppingToken 注册关闭回调
}
}行为要点：
| 行为 | 取值 | 含义 |
|---|---|---|
autoAck | false | 手动确认，处理失败可重试 |
prefetchCount | 1 | 单个消费者一次只取一条，保证公平分发 |
| 失败处理 | BasicNack(requeue: true) | 消息重回队列，可能无限重试 |
| 连接失败 | return | 消费者线程静默退出，不抛 BackgroundService 异常（BackgroundServiceExceptionBehavior = Ignore） |
六、消费者数量配置
RabbitMqConsumerOptions.ConsumerCount = N
│
├─ N=1 单线程串行消费（默认，保证顺序）
├─ N>1 多 BackgroundService 并行消费（吞吐↑，顺序不保证）
└─ N=0 一个消费者都不注册重要前提：AddHostedService<RabbitMqConsumer>() 注册 N 个实例，但每个实例构造时注入的 IMessageHandler 是容器里同一份默认解析结果（最后一个注册的 handler）。也就是说多消费者不会自动分担不同的队列，而是同一个 handler 被 N 个 BackgroundService 共用同一 channel。
实操建议：
- 吞吐优先：单一热点队列 →
ConsumerCount调大 2~4，但要保证 handler 幂等（因为乱序 + 可能重复投递） - 多队列：
ConsumerCount只能对着一个 handler 加并发，多队列场景建议自己在 handler 内做路由分发，或扩展AddRabbitMQ支持多 handler - 严格顺序：保持
ConsumerCount = 1
# 修改后重启生效
"RabbitMqConsumerOptions": { "ConsumerCount": 4 }七、与事件总线存储器的区别
| 维度 | RabbitMQ 业务队列（本篇） | RabbitMQ 作为事件源存储器 |
|---|---|---|
| 配置 | 同上开关 EventBus:EventSourceType=RabbitMQ | 同上开关 |
| 写入方式 | _producer.SendMessage(queue, msg) | _eventPublisher.PublishAsync(eventId, payload) |
| 队列名 | 业务自定义，如 order_created | 固定 adminnet_eventsource_queue（Startup.cs） |
| 消费方式 | IMessageHandler.HandleMessageAsync | [EventSubscribe(eventId)] 标记的方法 |
| 消费位置 | 本进程 BackgroundService | 本进程 IEventSubscriber（跨实例竞争同一队列） |
| 重试 | BasicNack(requeue:true) 无限重投 | RetryEventHandlerExecutor（NumRetries / RetryTimeout） |
| 典型用途 | 跨服务业务消息、削峰 | 写日志、发邮件/短信/企微（进程内订阅必须在线） |
事件总线切到 RabbitMQ 的注册代码（Admin.NET.Web.Core/Startup.cs）：
services.AddEventBus(options =>
{
options.UseUtcTimestamp = false;
options.LogEnabled = false;
options.AddExecutor<RetryEventHandlerExecutor>();
options.UnobservedTaskExceptionHandler = (obj, args) =>
{
if (args.Exception?.Message != null)
Log.Error($"事件总线未处理异常 ：{args.Exception?.Message} ", args.Exception);
};
var eventSourceType = App.GetConfig<string>("EventBus:EventSourceType", true);
if (eventSourceType == "Redis")
{
if (App.GetConfig<CacheOptions>("Cache", true).CacheType == CacheTypeEnum.Redis.ToString())
options.ReplaceStorerOrFallback(sp => new RedisEventSourceStorer(
sp.GetRequiredService<NewLife.Caching.ICacheProvider>(), "adminnet_eventsource_queue", 3000));
}
else if (eventSourceType == "RabbitMQ")
{
var o = App.GetConfig<RabbitMqOptions>("RabbitMQ");
var storer = new RabbitMQEventSourceStore(new ConnectionFactory
{
UserName = o.UserName, Password = o.Password,
HostName = o.HostName, Port = o.Port, VirtualHost = o.VirtualHost,
}, "adminnet_eventsource_queue", 3000);
options.ReplaceStorerOrFallback(_ => storer);
}
});Redis 分支有个隐藏前提：只有
Cache.CacheType == Redis时才会替换存储器，否则静默降级为内存队列。切模型前先确认缓存配置。
八、步骤清单
- 部署 RabbitMQ，建 vhost 与账号（注意 RabbitMQ 3.x 起
guest只能本机登录） MessageQueue.json里EventBus.EventSourceType改成"RabbitMQ"- 填对
RabbitMQ节点：HostName / Port / UserName / Password / VirtualHost - 写
XxxHandler : IMessageHandler（指定QueueName，实现HandleMessageAsync） - 业务处注入
RabbitMqProducer，调用SendMessage(queueName, json) - 按需调
RabbitMqConsumerOptions.ConsumerCount - 重启应用；控制台出现
[x] Received from xxx: ...即为通路 - 到
/kapi（Knife4j）确认只是 demo；真实消费观察日志表的日常日志
⚠️ 注意事项 / 常见坑
- 没改
EventBus:EventSourceType等于没接：AddRabbitMQ第一行就 return，生产会发现消息发出去了 RabbitMQ 里没有（其实连接都没建）。 - 多
IMessageHandler实现只有最后一个生效：RabbitMqConsumer构造注入单个IMessageHandler，DI 取最后注册者。想多队列请显式扩展或用不同的Register顺序配合IEnumerable<IMessageHandler>。 - 失败会无限重投：
requeue: true无重试上限、无死信交换机，一条毒丸消息会一直刷错误日志。建议 handler 内部做重试计数 + 落异常表 + 主动吞掉异常（但要小心丢消息）。 - 消息幂等必须自己做：手动 ACK + requeue 意味着同一条消息可能被投递多次，务必用幂等键（如
sys_idempotent_lock:或业务唯一索引）去重。 - 后台线程不能用作用域仓储：
BackgroundService长期存活，SqlSugarRepository<T>（Scoped）会被释放。统一写法是IServiceScopeFactory.CreateScope()+GetRequiredService<ISqlSugarClient>().CopyNew()。 - 连接失败是静默退出：不会抛到宿主。建议加健康检查或看首启日志里的
RabbitMQ连接失败，请检查配置文件。 - 只支持默认交换机：
exchange: string.Empty，要做发布/订阅或路由模式需要自行扩展RabbitMqConnection/ 新的 Producer。 mandatory: true但没监听退回事件：目标队列不存在时消息会被 Broker 回退而代码没订阅BasicReturnAsync，表现为"静默丢失"。确保先QueueDeclare（ Producer 已做）再发。ConsumerCount是字符串："4"而非4，写错类型会取默认值。- 别把大数据塞队列：任意 Broker 都不适合传 MB 级 payload，传对象存储 URL/ID 即可。
- 和 SignalR/事件总线共用 Redis 时要注意 db 隔离：
VirtualHost与 RedisdefaultDatabase是两套体系，别指望它们互通。
小结
- 打开开关：
MessageQueue.json→EventBus.EventSourceType = "RabbitMQ"，并在Startup.cs里services.AddRabbitMQ(consumerCount) - 生产：
RabbitMqProducer.SendMessage(queueName, message)，默认交换机 + durable + persistent - 消费：实现
IMessageHandler（QueueName+HandleMessageAsync），自动注册 - 消费语义：手动 ACK、
prefetch=1、失败BasicNack(requeue:true)无限重投 → 必须做幂等 - 消费者数量由
RabbitMqConsumerOptions.ConsumerCount控制，多消费者不保证顺序 - 与事件总线存储器共用连接配置，但队列、消费方式、重试机制完全不同，别混用