为何向RabbitMQ队列发布100万条消息总是存在缺失？

可用于追踪缺失消息的额外事件 #

除了你已订阅的事件，以下事件和机制能帮助定位消息丢失原因：

1. Connection.Shutdown 事件 #

当RabbitMQ连接异常或正常关闭时触发，可获取关闭的具体原因（如网络中断、节点重启、权限问题等）。示例代码：

connection.Shutdown += (sender, args) =>
{
    writer.WriteLine($"Connection shutdown: {args.ReplyText}, Reason: {args.ShutdownReason} - {DateTime.Now.ToString("hh:mm:ss:ffffff")}");
};

2. Channel.CallbackException 事件 #

捕获通道异步操作（如消息确认、返回）中未被主线程try-catch捕获的异常，这类静默异常可能导致消息处理逻辑失败。示例代码：

channel.CallbackException += (sender, args) =>
{
    writer.WriteLine($"Channel callback exception: {args.Exception.Message} - {DateTime.Now.ToString("hh:mm:ss:ffffff")}");
};

3. Channel.ModelShutdown 事件 #

当通道被关闭时触发，可获取通道关闭的详细原因，比如通道因错误被RabbitMQ强制关闭。示例代码：

channel.ModelShutdown += (sender, args) =>
{
    writer.WriteLine($"Channel shutdown: {args.ReplyText}, Reason: {args.ShutdownReason} - {DateTime.Now.ToString("hh:mm:ss:ffffff")}");
};

代码中的关键问题与修复建议 #

你的代码存在几个可能导致消息丢失的核心问题：

1. 未等待所有消息确认 #

虽然开启了ConfirmSelect()启用发布确认模式，但发布完消息后直接结束程序，未等待RabbitMQ返回所有消息确认。此时连接和通道被立即释放，缓冲区中未发送或未确认的消息会被丢弃。

修复：发布完所有消息后添加等待确认逻辑：

// 等待所有消息确认，超时时间设为5分钟
if (!channel.WaitForConfirms(TimeSpan.FromMinutes(5)))
{
    writer.WriteLine("部分消息未在超时时间内得到确认 - " + DateTime.Now.ToString("hh:mm:ss:ffffff"));
}
// 或者使用WaitForConfirmsOrDie，只要有消息未确认就抛出异常
// channel.WaitForConfirmsOrDie(TimeSpan.FromMinutes(5));

2. StreamWriter未正确释放 #

第一个StreamWriter未用using块包裹，可能导致日志未及时写入磁盘，丢失关键追踪信息。修复：

using (StreamWriter writer = new StreamWriter("PATH", append: true))
{
    writer.WriteLine(DateTime.Now.ToString("hh:mm:ss:ffffff"));
    
    // 后续所有日志写入、连接创建、消息发布逻辑都放在此块内
    
    writer.WriteLine(DateTime.Now.ToString("hh:mm:ss:ffffff"));
}

3. 缺乏消息计数校验 #

添加发布计数和确认计数统计，对比差值确认丢失阶段：

int publishedCount = 0;
long ackedCount = 0;

channel.BasicAcks += (sender, eventArgs) =>
{
    ackedCount += eventArgs.DeliveryTag - ackedCount; // 处理RabbitMQ批量确认
    writer.WriteLine($"已确认消息数: {ackedCount} - {DateTime.Now.ToString("hh:mm:ss:ffffff")}");
};

foreach (Message message in messages)
{
    byte[] payload = Encoding.UTF8.GetBytes(message.Content.Data);
    channel.BasicPublish("my_exchange", "my.routingkey.publish", mandatory: true, properties, payload);
    publishedCount++;
}

writer.WriteLine($"总发布数: {publishedCount}, 总确认数: {ackedCount} - {DateTime.Now.ToString("hh:mm:ss:ffffff")}");

其他排查方向 #

• 检查RabbitMQ节点日志：查看节点日志文件，确认是否有消息被拒绝、队列溢出、磁盘空间不足等错误。
• 确认交换器与队列绑定：确保my_exchange和路由键my.routingkey.publish正确绑定到目标队列，且队列未设置过期或长度限制。
• 调整客户端参数：修改ConnectionFactory的RequestedFrameMax和RequestedHeartbeat参数，避免缓冲区溢出或心跳超时导致消息丢失。

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