开发者社区> 凯京技术团队> 正文

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

简介: RabbitMQ通过HTTP API获取队列消息内容 背景 为配合公司数据中台建设,我们对公司内部使用的mysql binlog数据分发中间件(现已开源:https://gitee.com/kekingcn/keking-binlog-distributor )进行定制化改造,使其能够同时支持re.
+关注继续查看

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

背景

为配合公司数据中台建设,我们对公司内部使用的mysql binlog数据分发中间件(现已开源: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 ,希望我们的开源项目能反哺社区,为大家工作带来便利。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
开源项目推荐:物联网消息引擎EMQ X/大数据平台TDengine/可视化平台Grafana
开源项目推荐:物联网消息引擎EMQ X/大数据平台TDengine/可视化平台Grafana
150 0
RabbitMQ通过HTTP API获取队列消息内容
RabbitMQ通过HTTP API获取队列消息内容 背景 为配合公司数据中台建设,我们对公司内部使用的mysql binlog数据分发中间件(现已开源:https://gitee.com/kekingcn/keking-binlog-distributor )进行定制化改造,使其能够同时支持re.
3364 0
微信开发之获取openid及推送模板消息
有很多的朋友再问我怎么获取code,openid之类的问题,在这里我就给大家分享一下。 在做微信支付是需要获取openid的,推送模板消息也是需要openid包括其他一些功能分享等也都是需要的,openid是什么呢,这里给大家解释一下,是用户在公众号中的一个标识,就相当于你数据表中的ID一样,一个身份标识。
3268 0
物联网平台实用技巧:调用API获取设备状态
物联网很多业务场景中,时常需要获取设备的实时状态,以便根据不同状态(在线或离线)做不同处理。阿里云物联网平台提供多个云端API来获取设备的状态信息。本文介绍这些API的调用方法。
16104 0
通过HTTP/2通道实时获取IoT设备状态和数据
通过HTTP/2通道实时获取IoT设备状态和数据 1.服务端订阅 1.1 服务端订阅流程 在IoT场景,有时候我们期望业务服务器能接收到设备状态和设备采集的数据,而不是通过云产品中转, 这时我们可以开启服务端订阅,IoT平台会把设备产生的消息通过HTTP/2通道推送到业务服务器,以便根据自身业务场景消费。
2389 0
赠2000万次 API 请求次数 | 消息队列MQ限时特惠
消息队列MQ推出新用户体验尝鲜包: 赠送增2000万次 API 请求次数 150次Topic规格,有效时长达6个月 限时特惠6.8折(降价前270元) -> 特惠入口
1082 0
【计算机网络】HTTP 与 HTTPS ( HTTPS 简介 | HTTP 通信过程 )
【计算机网络】HTTP 与 HTTPS ( HTTPS 简介 | HTTP 通信过程 )
33 0
阿里云微服务消息队列MQTT管控API本地测试Quick Start
微消息队列MQTT版是阿里云推出的一款面向移动互联网以及物联网领域的轻量级消息中间件。如果说传统的消息队列中间件一般应用于微服务之间,那么适用于物联网的微消息队列MQTT版则实现了端与云之间的消息传递和真正意义上的万物互联。目前产品提供常用的云端API的功能调用,本文通过接口调用QuerySessionByClientId根据Client ID查询指定设备的连接信息。
170 0
+关注
凯京技术团队
我们是凯京科技技术团队。技术有无穷的魅力,我们不满足当前的现状,脚踏实地锐意进取,定能触碰到星空
24
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
JS零基础入门教程(上册)
立即下载
性能优化方法论
立即下载
手把手学习日志服务SLS,云启实验室实战指南
立即下载