常见问题排查

本文介绍RabbitMQ生产环境常见问题的排查思路与处置方案。

定义

常见问题排查是针对消息积压、连接泄漏、内存告警、队列阻塞等生产故障，通过指标采集、日志分析与工具诊断，快速定位根因并恢复服务的标准化流程。

原理

问题分类与指标

排查路径

告警触发
    │
    ├── 消息积压 ──▶ 检查消费者状态 ──▶ 分析消费TPS ──▶ 扩容/修复
    │
    ├── 连接泄漏 ──▶ 分析connection列表 ──▶ 定位异常IP ──▶ 清理/修复
    │
    ├── 内存告警 ──▶ 查看memory breakdown ──▶ 检查binary ──▶ GC/重启
    │
    └── 磁盘告警 ──▶ 查看持久化队列 ──▶ 清理过期消息 ──▶ 扩容磁盘

示例

Maven依赖

<dependencies>
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.20.0</version>
    </dependency>
</dependencies>

消息积压排查

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import java.io.IOException;

public class MessageBacklogCheck {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);

        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            // 被动声明获取队列信息
            com.rabbitmq.client.AMQP.Queue.DeclareOk ok =
                ch.queueDeclarePassive("order_queue");
            System.out.println("Messages ready: " + ok.getMessageCount());
            System.out.println("Consumers: " + ok.getConsumerCount());

            // 采样查看消息内容
            for (int i = 0; i < 5; i++) {
                GetResponse resp = ch.basicGet("order_queue", true);
                if (resp != null) {
                    String body = new String(resp.getBody());
                    System.out.println("Sample msg: " + body);
                }
            }
        } catch (IOException e) {
            System.err.println("Check failed: " + e.getMessage());
        }
    }
}

连接泄漏排查

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Address;
import java.util.ArrayList;
import java.util.List;

public class ConnectionLeakCheck {
    public static void main(String[] args) throws Exception {
        // 模拟连接泄漏场景
        List<Connection> connections = new ArrayList<>();
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        // 创建多个连接但不关闭
        for (int i = 0; i < 20; i++) {
            Connection conn = factory.newConnection();
            connections.add(conn);
            System.out.println("Created connection: " + conn.getClientProvidedName());
        }

        // 正确做法：使用try-with-resources或显式关闭
        System.out.println("--- Proper cleanup ---");
        for (Connection conn : connections) {
            conn.close();
        }
        System.out.println("All connections closed");
    }
}

诊断命令

# 查看队列详情
rabbitmqadmin list queues name messages messages_ready messages_unacknowledged consumers

# 查看连接列表与空闲时间
rabbitmqadmin list connections name peer_host channels state idle_since

# 查看内存分布
rabbitmqctl report | grep -A 20 "Memory breakdown"

# 查看阻塞队列
rabbitmqctl list_queues name consumers state | grep -E "^[\w-]+\s+0\s+"

# 强制关闭空闲连接
rabbitmqctl close_connection "<connection_pid>" "idle connection cleanup"

消息积压应急消费

import com.rabbitmq.client.*;

public class EmergencyConsumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        // 应急模式：提高prefetch加速消费
        factory.setRequestedChannelMax(200);

        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            ch.basicQos(1000); // 加大prefetch
            ch.basicConsume("order_queue", true, new DefaultConsumer(ch) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                        AMQP.BasicProperties properties, byte[] body) {
                    // 快速消费不阻塞
                    processMessage(new String(body));
                }
            });
            Thread.sleep(60000); // 应急运行1分钟
        }
    }

    private static void processMessage(String msg) {
        // 简化处理逻辑
    }
}

注意事项

消息积压时不要盲目重启消费者，可能触发大量requeue导致雪崩。

连接泄漏排查优先检查客户端代码是否正确关闭连接，尤其是异常分支。

内存告警达到high watermark时Broker会阻塞生产者，优先排查Refc binary。

磁盘告警时Broker会拒绝写入，需提前设置监控阈值避免服务中断。

排查期间开启tracing会影响性能，仅在必要时临时启用。

要点总结

• 问题分类：消息积压、连接泄漏、内存/磁盘告警、队列阻塞
• 排查路径：指标采集 -> 根因定位 -> 应急处置 -> 修复验证
• 连接泄漏优先检查客户端关闭逻辑，尤其异常分支
• 内存告警重点排查Refc binary与持久化消息堆积
• 应急消费可提高prefetch加速，但需避免消费者雪崩