RabbitMQ监控

1、搜集日志信息

通过ELK搜集集群节点中的日志信息，对于ERROR信息，进行邮件告知相关的维护人员及时进行处理。

2、监控硬件资源及进程

公司已经有监控系统，直接将如下磁盘、内存及端口、URL信息在监控中进行配置即可。

磁盘

监控每台服务器的磁盘空间，重点监控如下两个目录：

• log目录

即/app/log 目录。日志文件可以通过定期压缩清理，保证控制在一定的大小内。

• data目录

即/app/data 目录，这个目录主要堆积的数据还是未消费的已经持久化到磁盘上的消息。这个需要及时对消息进行监控。

内存

监控服务器的内存使用情况，当达到阀值的情况下，进行提前告警。

监控进程

主要是监控5672端口，及http://xxxx:15672访问地址

3、通过rest接口进行监控

依托于rabbitmq提供的强大的rest api，我们可以很方便的获取到集群中的各个节点信息，从而根据具体的需求，进行全方位的监控。

在公司已经采购的监控系统和自行搭建的zabbix监控中，都能够很好的支持rest api。可以通过监控系统对json结果的解析，达到告警的目的。

消息监控

消息监控可以通过定时调用rest接口，获取队列的未消费数量，对超过一定数量的mq消息，提供短信邮件告警；

• 获取消息的rest接口地址为：

http://xxx:15672/api/queues

• 消息内容示例（只截取部分内容）：

[{
    "memory": 55568,
    ...
    "messages": 21,                              #消息总数量
    ...
    "messages_ready": 21,                        #待处理消息数量
    ...
    "messages_unacknowledged": 0,                #未确认消息数量
    ...
    "idle_since": "2018-07-02 14:02:04",
    "consumer_utilisation": null,
    "policy": "hao123",
    "exclusive_consumer_tag": null,
    "consumers": 0,
    "slave_nodes": ["rabbit@rabbitmqstg03", "rabbit@rabbitmqstg04"],
    "synchronised_slave_nodes": ["rabbit@rabbitmqstg03", "rabbit@rabbitmqstg04"],
    "recoverable_slaves": ["rabbit@rabbitmqstg03", "rabbit@rabbitmqstg04"],
    "state": "running",
    ...
    "message_bytes_ram": 13086,                  #消息占用的内存大小
    "message_bytes_persistent": 2816,            #消息占用的磁盘大小
    ...
    "name": "ha.queue_123",
    "vhost": "/123",
    "durable": true,
    "auto_delete": false,
    "exclusive": false,
    "arguments": {},
    "node": "rabbit@rabbitmqstg02"
}]

文件描述符、socket文件描述符、erlang进程监控

通过rest接口获取到每个集群中的节点信息，通过实际值和阀值的大小，及时的进行告警，这一步告警方式，可以根据严重程度，依次邮件、微信、短信通知到相关的人员。

• 获取节点的rest接口地址为：

http://xxx:15672/api/nodes

• 集群节点内容示例（只截取部分内容）：

[{
    "cluster_links": [{
        "peer_addr": "10.1.110.39",
        "peer_port": 51715,
        "sock_addr": "10.1.110.37",
        "sock_port": 25672,
        "stats": {...},
        "name": "rabbit@rabbitmqstg04"
    }, ...],
    "disk_free": 28271468544,                 # 磁盘使用量
    "fd_used": 58,                            # 文件描述符使用数
    ...
    "mem_used": 58362720,                     # 内存使用量
    "mnesia_disk_tx_count": 7,
    "mnesia_ram_tx_count": 13108,
    "proc_used": 225,                         # erlang进程使用数
    "sockets_used": 0,                        # socket描述符已经使用数
    "fd_total": 65536,                        # 文件描述符总数
    "sockets_total": 58890,                   # socket描述符总数
    "mem_limit": 1607527628,                  # 内存阀值
    "disk_free_limit": 10000000000,           # 磁盘阀值
    "proc_total": 1048576,                    # erlang进程总数
    ...
    "log_file": "/app/rabbitmq/rabbitmq_server-3.6.1/log/rabbitmq/rabbit@rabbitmqstg02.log",
    "sasl_log_file": "/app/rabbitmq/rabbitmq_server-3.6.1/log/rabbitmq/rabbit@rabbitmqstg02-sasl.log",
    "db_dir": "/app/rabbitmq/rabbitmq_server-3.6.1/data/rabbit@rabbitmqstg02",
    "config_files": ["/app/rabbitmq/rabbitmq_server-3.6.1/etc/rabbitmq/rabbitmq.config"],
    ...
    "name": "rabbit@rabbitmqstg02",
    "type": "disc",
    "running": true
}]

4、监听堵塞消息

可以在MQ中专门创建一个监控的队列，定时的发送和消费队列中的消息，并且通过的如下的代码去监控队列是否已经堵塞，如果监听到队列已经堵塞，就立即发送告警的短信和邮件。

关于是否需要手动将故障机器剔除集群，这个有待商榷，但是如果要做的话，是可以在这里做到的。

ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
connection.addBlockedListener(new BlockedListener() {
    public void handleBlocked(String reason) throws IOException {
        // Connection is now blocked
    }

    public void handleUnblocked() throws IOException {
        // Connection is now unblocked
    }
});