RabbitMQ通过HTTP API获取队列消息内容

谁借莪１个温暖的怀抱￠ 2022-02-12 22:37 250阅读 0赞

# RabbitMQ通过HTTP API获取队列消息内容 #

## 背景 ##

为配合公司数据中台建设，我们对公司内部使用的mysql binlog数据分发中间件（现已开源：[https://gitee.com/kekingcn/keking-binlog-distributor][https_gitee.com_kekingcn_keking-binlog-distributor] ）进行定制化改造，使其能够同时支持redis和rabbitmq两个分发渠道，并能在前端页面监控队列长度、队列消息内容等信息

## 需求 ##

由于之前已经完成后端分发功能、redis队列监控的开发，目前需要前端同时支持rabbitmq队列监控，并需要能够预览队列中的内容

## 方案选择 ##

要在前端实现显示rabbitmq队列及队列内容，有如下两种实现方案：

*  后端主服务使用rabbitmq客户端，使用amqpadmin读取当前vhost下队列列表，注册消费方，消费队列且不发送ACK确认，让其重新入队列
 *  后端主服务通过rabbitmq提供的HTTP API直接获取队列列表，和队列中的内容（rabbitmq前台管理页面就是通过HTTP API获取的队列列表和获取队列内消息）

对比两种方案，发现前者存在两点劣势：

1.  主服务需要额外依赖rabbitmq客户端，且ampqadmin需要有管理员权限的账号和密码
2.  主服务需要和业务系统同时消费队列数据，编码不当可能给整个系统带来稳定性问题，且消费数据不发送ACK确认应答不太合理

而且作为对比，HTTP API只需要能登陆rabbitmq管理页面的权限，并且官方前台管理页面已经实现了获取消息功能，不用自己消费队列数据

综上分析，选择使用HTTP API方式获取队列及队列内容更为合理

## 难点分析 ##

*  rabbitmq的HTTP API认证方式不明，通过浏览器调试查看官方前台管理页面http请求并不能看出明确的认证信息
 *  rabbitmq的HTTP API获取消息接口获取到的数据是base64编码的字符串，我们在消息中传输的是java实体对象，base64编码转二进制流再反序列化成java实体对象过程可能有障碍。api返回数据如下图所示
    
        ![image](https://wx4.sinaimg.cn/large/7dfa0a7bly1g1bffv5ur2j21670u046o.jpg)

## 方案验证 ##

1.  rabbitmq的HTTP API的认证问题可以通过其官方提供的http-client包解决，只需要添加如下依赖
    
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>http-client</artifactId>
            <version>3.1.1.RELEASE</version>
        </dependency>
    
    然后使用com.rabbitmq.http.client.Client构造方法传入如下apiUrl, username, password三个参数
    
        new Client(url, username, password);
    
    此client中的http请求对象 `private final RestTemplate rt;`就可以发送http请求并通过认证
2.  base64转二进制byte\[\]可以直接用apache-common-codec包直接解码，依赖如下
    
        <dependency>
            <groupId>commons-codec</groupId>
            <artifactId>commons-codec</artifactId>
            <version>1.12</version>
        </dependency>
    
    直接调用`org.apache.commons.codec.binary.Base64.decodeBase64(String base64String)`即可将base64编码的字符串转成二进制byte\[\]数组，至于反序列化，只需要保证和rabbitmq序列化过程对应即可将二进制字节码反序列化为java实体对象

## 编码实现 ##

经过方案验证，编码实现就非常简单了，具体实现如下，重写com.rabbitmq.http.client.Client类，添加如下方法

/**
      * @param vhost Virtrul Host
      * @param queuename 队列名
      * @param count 要取的队列消息数
      * @return EventBaseDTO实体列表
      */
    public List<EventBaseDTO> getMessageList(String vhost, String queuename, long count) throws IOException, ClassNotFoundException {
        String bodyStr = "{\"count\":" + count + ",\"requeue\":true,\"encoding\":\"base64\"}";
        Map<String, String> body = JSON.parseObject(bodyStr, Map.class);
        final URI uri = uriWithPath("./queues/" + encodePathSegment(vhost) + "/" + encodePathSegment(queuename) + "/get");
        ResponseEntity<List> result= rt.postForEntity(uri, body, List.class);
        List<EventBaseDTO> resultList = new ArrayList<>();
        List list = result.getBody();
        for (Object entry : list) {
            if (entry instanceof Map) {
                String base64Str = ((Map) entry).get("payload") == null ? "" : ((Map) entry).get("payload").toString();
                byte[] bytes = Base64.decodeBase64(base64Str);
                ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
                ObjectInputStream ois = new ObjectInputStream(bais);
                EventBaseDTO dto = (EventBaseDTO) ois.readObject();
                resultList.add(dto);
            }
        }
        return resultList;
    }

## 实现效果 ##

后端提供接口后，完成前端开发，即可实现同时对redis和rabbitmq队列的监控，效果如下图所示

![image][]

此次改进很快也会同步到开源社区版本 [https://gitee.com/kekingcn/keking-binlog-distributor][https_gitee.com_kekingcn_keking-binlog-distributor] ，希望我们的开源项目能反哺社区，为大家工作带来便利。

[https_gitee.com_kekingcn_keking-binlog-distributor]: https://gitee.com/kekingcn/keking-binlog-distributor
[image]: /images/20220213/56b1276baee64019a1ac4e01abfa8b70.png