摘要：任何没有监控的系统上线，一旦在生产环境发生故障，那么排查和修复问题的及时性将无法得到保证

一、为何要对消息中间件进行监控？

上线的业务系统需要监控，然而诸如消息队列、数据库、分布式缓存等生产环境的中间件系统也同样需要监控，否则一旦出现任何故障，排查和修复起来的时间和投入的人力成本都会大大增加，同时也不易利于日后进行问题原因的总结和复盘。
对于消息中间件RabbitMQ集群来说，没有监控能力更是灾难性的。假设，运行在生产环境上的中小规模RabbitMQ集群（总共有30个节点），出现部分节点服务不可用而导致生产者和消费者的业务工程无法正常发布和订阅消息（比如，Erlang虚拟机内存即将达到上限，服务器磁盘容量不足或者队列出现大量消息堆积），此时通过前置软负载HAProxy的Web页面又无法快速定位和排查出根本问题所在，那么研发和运维同事唯有对每台服务器的日志进行排查才可能发现并解决存在的问题。这种方式耗时又耗力，在生产环境对于故障响应时间和解决时间都非常重要，因此非常有必要对诸如像RabbitMQ这样的消息中间件进行各种参数的监控。

二、如何对RabbitMQ集群监控？

（1）RabbitMQ自带的Web管理端的插件

RabbitMQ作为一款在金融领域应用非常成熟的消息中间件,必然少不了监控功能，RabbitMQ提供了Web版的页面监控（只在本地的浏览器端访问地址：http://xxx.xxx.xxx.xxx:15672/，默认端口号是15672）。Web监控页面如下图所示：

RabbitMQ的Web监控截图.jpg
RabbitMQ的Web监控截图2.jpg
当然想要使用上面的RabbitMQ自带的Web监控页面，必须开启rabbitmq_management插件模式，需要在部署RabbitMQ实例的服务器的sbin文件夹下执行如下命令：

#先启用rabbitmq_management插件
./rabbtimq-plugins enable rabbitmq_management
#然后停止MQ服务再重启
rabbitmqctl stop
rabbitmq-server -detached

如果只是在测试或生产环境小规模地应用RabbitMQ消息队列（比如业务并发访问量较低），那么简单地用用RabbitMQ自带的Web页面进行监控也就足够了。但是，如果对RabbitMQ的并发性能、高可用和其他参数都有一些定制化的监控需求的话，那么我们就有必要采用其他的方式来达到该目标。

（2）RabbitMQ的tracing消息轨迹追踪

对于金融级或者工业级应用场景来说，消息收发的可靠性永远是排在第一位的。消息队列集群可能因为各种问题（比如，生产者/消费者与RabbitMQ的服务器断开连接、Erlang虚拟机挂了、消息积压导致RabbitMQ内存达到最大阀值），难免会出现某条消息异常丢失或者客户端程序无法发送接收消息的情况。因此，这个时候就需要有一个较好的机制跟踪记录消息的投递过程，以此协助开发和运维人员进行问题的定位。

(a)RabbitMQ的tracing原理

RabbitMQ自带的tracing Log插件功能可以完成对于集群中各个消息投递/订阅的轨迹进行追踪。RabbitMQ tracing log的原理是将生产者投递给RabbitMQ服务器的消息，或者RabbitMQ服务器投递给消费者的消息按照指定格式发送至默认的交换器上。这个默认的交换器名称为“amq.rabbitmq.trace”，是一个topic类型的交换器。随后RabbitMQ会创建一个绑定了这个交换器的队列amq.gen队列。通过这个交换器，把消息的流入和流出情况进行封装后发送到amq.gen队列中，该队列会把消息流转的日志记录在相应的日志中。

(b)启用RabbitMQ tracing来消息追踪

这里可以使用rabbitmq-plugins enable rabbitmq_tracing命令来启动rabbitmq_tracing插件。

[root@mq-testvm1 sbin]# rabbitmq-plugins enable rabbitmq_tracing
The following plugins have been enabled:
  rabbitmq_tracing

Applying plugin configuration to rabbit@rmq-broker-test-1... started 1 plugin.

其对应的关闭插件的命令是：rabbitmq-plugins disable rabbitmq_tracing
在Web管理界面 “Admin”右侧原本只有”Users”、”Virtual Hosts”以及”Policies“这个三Tab项，在添加rabbitmq_tracing插件之后，会多出”Tracing”这一项内容。同时，添加名称为“trace1”的消息追踪任务。

RabbitMQ启用tracing_log后的UI界面.jpg
在添加完trace之后，会根据匹配的规则将相应的消息日志输出到对应的trace文件之中，文件的默认路径为/var/tmp/rabbitmq-tracing。可以在页面中直接点击“Trace log files”下面的列表直接查看对应的日志文件。此外，在“Queues”队列一栏中可以看到又多了一个如下队列：
RabbitMQ添加完tracing_log后出现的队列.jpg
当通过Web UI页面发布一条消息后，对应的Tracing log的Text格式的消息日志参考如下：

================================================================================
2018-05-27 8:16:34:545: Message published

Node:         rabbit@rmq-broker-test-1
Connection:   <rabbit@rmq-broker-test-1.2.10776.3>
Virtual host: /
User:         root
Channel:      1
Exchange:     
Routing keys: [<<"pressure_1">>]
Routed queues: [<<"pressure_1">>]
Properties:   [{<<"delivery_mode">>,signedint,1},{<<"headers">>,table,[]}]
Payload: 
adfadfadf

（3）采用RabbitMQ的HTTP API接口进行监控

要构建独立的监控系统，可以使用RabbitMQ本身提供的Restful HTTP API接口来获取各种业务监控需要的实时数据。当然，这个接口的作用远不止于获取一些监控数据，也可以通过这些HTTP API来操作RabbitMQ进行各种集群元数据的添加/删除/更新的操作。
下面列举了可以利用RabbitMQ的HTTP API接口实现的各种操作：

HTTP API URL	HTTP 请求类型	接口含义
/api/connections	GET	获取当前RabbitMQ集群下所有打开的连接
/api/nodes	GET	获取当前RabbitMQ集群下所有节点实例的状态信息
/api/vhosts/{vhost}/connections	GET	获取某一个虚拟机主机下的所有打开的connection连接
/api/connections/{name}/channels	GET	获取某一个连接下所有的管道信息
/api/vhosts/{vhost}/channels	GET	获取某一个虚拟机主机下的管道信息
/api/consumers/{vhost}	GET	获取某一个虚拟机主机下的所有消费者信息
/api/exchanges/{vhost}	GET	获取某一个虚拟机主机下面的所有交换器信息
/api/queues/{vhost}	GET	获取某一个虚拟机主机下的所有队列信息
/api/users	GET	获取集群中所有的用户信息
/api/users/{name}	GET/PUT/DELETE	获取/更新/删除指定用户信息
/api/users/{user}/permissions	GET	获取当前指定用户的所有权限信息
/api/permissions/{vhost}/{user}	GET/PUT/DELETE	获取/更新/删除指定虚拟主机下特定用户的权限
/api/exchanges/{vhost}/{name}/publish	POST	在指定的虚拟机主机和交换器上发布一个消息
/api/queues/{vhost}/{name}/get	POST	在指定虚拟机主机和队列名中获取消息，同时该动作会修改队列状态
/api/healthchecks/node/{node}	GET	获取指定节点的健康检查状态

上面的HTTP API接口只是列举了RabbitMQ所支持的部分功能，读者可以参考RabbitMQ官方文档和访问http://server-name:15672/api/的Web页面来获取更多的其他接口信息。
业务研发的同学可以使用Apache的httpcomponents组件—HttpClient或者Spring的RestTemplate组件生成并发送HTTP的GET/POST/DELETE/PUT请求至RabbitMQ Server，根据自己的业务目标完成相应的业务监控需求。
下面是一个使用RabbitMQ Http API接口来获取集群监控参数的demo代码，主要使用HttpClient以及jackson来查询MQ集群的性能参数和存在的用户信息；

public class MonitorRabbitMQDemo {

    //RabbitMQ的HTTP API——获取集群各个实例的状态信息，ip替换为自己部署相应实例的
    private static String RABBIT_NODES_STATUS_REST_URL = "http://ip:15672/api/nodes";

    //RabbitMQ的HTTP API——获取集群用户信息，ip替换为自己部署相应实例的
    private static String RABBIT_USERS_REST_URL = "http://ip:15672/api/users";

    //rabbitmq的用户名
    private static String RABBIT_USER_NAME = "root";

    //rabbitmq的密码
    private static String RABBIT_USER_PWD = "root123";

    public static void main(String[] args) {
        try {
            //step1.获取rabbitmq集群各个节点实例的状态信息
            Map<String, ClusterStatus> clusterMap =
                    fetchRabbtMQClusterStatus(RABBIT_NODES_STATUS_REST_URL, RABBIT_USER_NAME, RABBIT_USER_PWD);

            //step2.打印输出各个节点实例的状态信息
            for (Map.Entry entry : clusterMap.entrySet()) {
                System.out.println(entry.getKey() + " : " + entry.getValue());
            }

            //step3.获取rabbitmq集群用户信息
            Map<String, User> userMap =
                    fetchRabbtMQUsers(RABBIT_USERS_REST_URL, RABBIT_USER_NAME, RABBIT_USER_PWD);

            //step2.打印输出rabbitmq集群用户信息
            for (Map.Entry entry : userMap.entrySet()) {
                System.out.println(entry.getKey() + " : " + entry.getValue());
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static Map<String,ClusterStatus> fetchRabbtMQClusterStatus(String url, String username, String password) throws IOException {
        Map<String, ClusterStatus> clusterStatusMap = new HashMap<String, ClusterStatus>();
        String nodeData = getData(url, username, password);
        JsonNode jsonNode = null;
        try {
            jsonNode = JsonUtil.toJsonNode(nodeData);
        } catch (IOException e) {
            e.printStackTrace();
        }
        Iterator<JsonNode> iterator = jsonNode.iterator();
        while (iterator.hasNext()) {
            JsonNode next = iterator.next();
            ClusterStatus status = new ClusterStatus();
            status.setDiskFree(next.get("disk_free").asLong());
            status.setFdUsed(next.get("fd_used").asLong());
            status.setMemoryUsed(next.get("mem_used").asLong());
            status.setProcUsed(next.get("proc_used").asLong());
            status.setSocketUsed(next.get("sockets_used").asLong());
            clusterStatusMap.put(next.get("name").asText(), status);
        }
        return clusterStatusMap;
    }

    public static Map<String,User> fetchRabbtMQUsers(String url, String username, String password) throws IOException {
        Map<String, User> userMap = new HashMap<String, User>();
        String nodeData = getData(url, username, password);
        JsonNode jsonNode = null;
        try {
            jsonNode = JsonUtil.toJsonNode(nodeData);
        } catch (IOException e) {
            e.printStackTrace();
        }
        Iterator<JsonNode> iterator = jsonNode.iterator();
        while (iterator.hasNext()) {
            JsonNode next = iterator.next();
            User user = new User();
            user.setName(next.get("name").asText());
            user.setTags(next.get("tags").asText());
            userMap.put(next.get("name").asText(), user);
        }
        return userMap;
    }

    public static String getData(String url, String username, String password) throws IOException {
        CloseableHttpClient httpClient = HttpClients.createDefault();
        UsernamePasswordCredentials creds = new UsernamePasswordCredentials(username, password);
        HttpGet httpGet = new HttpGet(url);
        httpGet.addHeader(BasicScheme.authenticate(creds, "UTF-8", false));
        httpGet.setHeader("Content-Type", "application/json");
        CloseableHttpResponse response = httpClient.execute(httpGet);

        try {
            if (response.getStatusLine().getStatusCode() != 200) {
                System.out.println("call http api to get rabbitmq data return code: " + response.getStatusLine().getStatusCode() + ", url: " + url);
            }
            HttpEntity entity = response.getEntity();
            if (entity != null) {
                return EntityUtils.toString(entity);
            }
        } finally {
            response.close();
        }

        return StringUtils.EMPTY;
    }

    public static class JsonUtil {
        private static ObjectMapper objectMapper = new ObjectMapper();
        static {
            objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
            objectMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
        }

        public static JsonNode toJsonNode(String jsonString) throws IOException {
            return objectMapper.readTree(jsonString);
        }
    }

    public static class User {
        private String name;
        private String tags;
        @Override
        public String toString() {
            return "User{" +
                    "name=" + name +
                    ", tags=" + tags +
                    '}';
        }
                //GET/SET方法省略
    }

    public static class ClusterStatus {
        private long diskFree;
        private long diskLimit;
        private long fdUsed;
        private long fdTotal;
        private long socketUsed;
        private long socketTotal;
        private long memoryUsed;
        private long memoryLimit;
        private long procUsed;
        private long procTotal;
        // 此处省略了Getter和Setter方法
        @Override
        public String toString() {
            return "ClusterStatus{" +
                    "diskFree=" + diskFree +
                    ", diskLimit=" + diskLimit +
                    ", fdUsed=" + fdUsed +
                    ", fdTotal=" + fdTotal +
                    ", socketUsed=" + socketUsed +
                    ", socketTotal=" + socketTotal +
                    ", memoryUsed=" + memoryUsed +
                    ", memoryLimit=" + memoryLimit +
                    ", procUsed=" + procUsed +
                    ", procTotal=" + procTotal +
                    '}';
        }
                //GET/SET方法省略
    }

运行上面的demo后可以看到输出的日志如下（demo中用httpclient仅仅为的是展示，真正开发中写的代码可以参考使用Spring RestTemplate，其为开发者进行了二次封装，可以一定程度提高开发效率）：

#输出测试环境所部署的10个节点的集群实例信息
rabbit@rmq-broker-test-8 : ClusterStatus{diskFree=34474188800, diskLimit=0, fdUsed=46, fdTotal=0, socketUsed=2, socketTotal=0, memoryUsed=383752384, memoryLimit=0, procUsed=1200, procTotal=0}
rabbit@rmq-broker-test-9 : ClusterStatus{diskFree=33215782912, diskLimit=0, fdUsed=34, fdTotal=0, socketUsed=0, socketTotal=0, memoryUsed=139520504, memoryLimit=0, procUsed=378, procTotal=0}
rabbit@rmq-broker-test-6 : ClusterStatus{diskFree=37309734912, diskLimit=0, fdUsed=45, fdTotal=0, socketUsed=0, socketTotal=0, memoryUsed=144497072, memoryLimit=0, procUsed=387, procTotal=0}
rabbit@rmq-broker-test-7 : ClusterStatus{diskFree=37314203648, diskLimit=0, fdUsed=45, fdTotal=0, socketUsed=0, socketTotal=0, memoryUsed=204729280, memoryLimit=0, procUsed=387, procTotal=0}
rabbit@rmq-broker-test-4 : ClusterStatus{diskFree=36212776960, diskLimit=0, fdUsed=45, fdTotal=0, socketUsed=0, socketTotal=0, memoryUsed=136635624, memoryLimit=0, procUsed=387, procTotal=0}
rabbit@rmq-broker-test-5 : ClusterStatus{diskFree=37313929216, diskLimit=0, fdUsed=46, fdTotal=0, socketUsed=2, socketTotal=0, memoryUsed=349737776, memoryLimit=0, procUsed=1206, procTotal=0}
rabbit@rmq-broker-test-2 : ClusterStatus{diskFree=37315076096, diskLimit=0, fdUsed=42, fdTotal=0, socketUsed=0, socketTotal=0, memoryUsed=173172688, memoryLimit=0, procUsed=381, procTotal=0}
rabbit@rmq-broker-test-3 : ClusterStatus{diskFree=37338624000, diskLimit=0, fdUsed=46, fdTotal=0, socketUsed=0, socketTotal=0, memoryUsed=88274008, memoryLimit=0, procUsed=387, procTotal=0}
rabbit@rmq-broker-test-1 : ClusterStatus{diskFree=36990242816, diskLimit=0, fdUsed=45, fdTotal=0, socketUsed=0, socketTotal=0, memoryUsed=180700296, memoryLimit=0, procUsed=409, procTotal=0}
rabbit@rmq-broker-test-10 : ClusterStatus{diskFree=33480851456, diskLimit=0, fdUsed=45, fdTotal=0, socketUsed=1, socketTotal=0, memoryUsed=126567928, memoryLimit=0, procUsed=792, procTotal=0}

#输出RabbitMQ集群中的用户元数据（包含用户名和Tag标签）
root : User{name=root, tags=administrator}
guest : User{name=guest, tags=administrator}

三、具备监控能力的RabbitMQ集群设计

上面介绍了三种不同的方式来对RabbitMQ集群进行监控，其实本质上来说，第一种和第三种方式是一致的，细心的同学会发现RabbitMQ的Web UI是定期执行刷行动作，向部署的实例发送HTTP GET/POST/PUT等相应的请求。
其中第一种能够监控的范围相对有限，更适合小众化地使用；第二种tracing log方式能够很好的监控消息投递和接收的轨迹，但是多少对集群性能有所损耗，在实际压测中发现这种方式会导致节点大量内存消耗，其生成的log日志也会影响磁盘的IO，因此只限于在开发和测试环境调试时使用；而第三种使用HTTP API监控则能够根据开发者的业务需求自定义监控范围，对于监控数据的精度也能够通过调整调用HTTP API的间隔来实现。因此，这里作者较为推荐使用第三种方式来对大规模的RabbitMQ集群进行监控。

RabbitMQ小规模集群的架构设计图(附加了监控部分).png

这里给出了带有监控功能的RabbitMQ集群架构设计图，对于集群部署的原理和软负载等内容都在上一篇《消息中间件—RabbitMQ（集群原理与搭建篇)》中有详细的阐述，图中作者自设计了一个MQ-Cluster-Agent工程用于监听RabbitMQ集群的状态，其中主要通过调用HTTP API接口来查询获取集群元数据。随后，每隔一定周期将这些监控数据push至Kafka集群中。后台的监控控制台工程可以使用Kafka stream流处理方式对Kafka消息队列中的准实时数据进行一定的业务加工，随后生成业务方需要的监控报表。

四、总结

本文主要详细介绍了为何需要对MQ消息中间件进行监控，以及监控RabbitMQ集群的三种主要方法，并最后给出了一种具备监控能力的RabbitMQ集群架构设计。限于篇幅原因，对于图中采用agent完成对集群进行准实时监控的设计方法以及使用Kafka完成流处理的方式将在后续的篇幅2中进行详细介绍。限于笔者的才疏学浅，对本文内容可能还有理解不到位的地方，如有阐述不合理之处还望留言一起探讨。