RabbitMQ通过HTTP API获取队列消息内容
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获取的队列列表和获取队列内消息)
对比两种方案,发现前者存在两点劣势:
- 主服务需要额外依赖rabbitmq客户端,且ampqadmin需要有管理员权限的账号和密码
- 主服务需要和业务系统同时消费队列数据,编码不当可能给整个系统带来稳定性问题,且消费数据不发送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)
方案验证
-
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请求并通过认证 -
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队列的监控,效果如下图所示
此次改进很快也会同步到开源社区版本 https://gitee.com/kekingcn/keking-binlog-distributor ,希望我们的开源项目能反哺社区,为大家工作带来便利。
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
浅析ServiceMesh & Istio
浅析ServiceMesh & Istio 前言 公司于18年10月正式确认服务容器化,到18年12月4日第一个服务正式部署到生产环境kubernetes集群,再到如今已有23个服务完成了生产环境容器化的切换,更多的服务在测试环境容器化部署随时可以切换到生产环境。目前新项目的开发,大部分都直接在测试环境容器化部署,不再需要新购ECS搭建测试环境。随着容器化的深入,服务间的通信和联系变的更加复杂,其中通信的可视化、流量的控制和服务质量的评估问题日益凸显,成为了微服务方案的短板。这个时候Service mesh就进入了我们的视野。 Service mesh是什么 Service mesh 又译作 “服务网格”,作为服务间通信的基础设施层。随着规模和复杂性的增长,服务网格越来越难以理解和管理。我们对它的需求包括服务发现、负载均衡、故障恢复、指标收集和监控,以及通更加复杂的运维需求,例如 A/B 测试、金丝雀发布、限流、访问控制和端到端认证等。它负责通过包含现代云原生应用程序的复杂服务拓扑来可靠地传递请求。实际上,服务网格通常通过一组轻量级网络代理来实现,这些代理与应用程序代码一起部署...
- 下一篇
Fescar&Seata分布式事务实现原理解析探秘
前言 fescar发布已有时日,分布式事务一直是业界备受关注的领域,fescar发布一个月左右便受到了近5000个star足以说明其热度。当然,在fescar出来之前,已经有比较成熟的分布式事务的解决方案开源了,比较典型的方案如LCN(https://github.com/codingapi/tx-lcn)的2pc型无侵入事务,目前lcn已发展到5.0,已支持和fescar事务模型类似的TCX型事务。还有如TCC型事务实现hmily(https://github.com/yu199195/hmily)、tcc-transaction(https://github.com/changmingxie/tcc-transaction)等。在微服务架构流行的当下、阿里这种开源大户背景下,fescar的发布无疑又掀起了研究分布式事务的热潮。fescar脱胎于阿里云商业分布式事务服务GTS,在线上环境提供这种公共服务其模式肯定经受了非常严苛的考验。其分布式事务模型TXC又仿于传统事务模型XA方案,主要区别在于资源管理器的定位一个在应用层一个在数据库层。博主觉得fescar的txc模型实现非常有研究...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
-
Docker使用Oracle官方镜像安装(12C,18C,19C)
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS8编译安装MySQL8.0.19
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- MySQL8.0.19开启GTID主从同步CentOS8
- CentOS7,8上快速安装Gitea,搭建Git服务器
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
推荐阅读
最新文章
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- CentOS6,CentOS7官方镜像安装Oracle11G
- SpringBoot2整合Redis,开启缓存,提高访问速度
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- SpringBoot2整合MyBatis,连接MySql数据库做增删改查操作
- Hadoop3单机部署,实现最简伪集群
- MySQL8.0.19开启GTID主从同步CentOS8
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果