RabbitMQ通过HTTP API获取队列消息内容 |
您所在的位置:网站首页 › rabbitmq队列长度查询 › 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返回数据如下图所示![]() 直接调用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 getMessageList(String vhost, String queuename, long count) throws IOException, ClassNotFoundException { String bodyStr = "{\"count\":" + count + ",\"requeue\":true,\"encoding\":\"base64\"}"; Map body = JSON.parseObject(bodyStr, Map.class); final URI uri = uriWithPath("./queues/" + encodePathSegment(vhost) + "/" + encodePathSegment(queuename) + "/get"); ResponseEntity result= rt.postForEntity(uri, body, List.class); List 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队列的监控,效果如下图所示
陈精华,2018年8月加入凯京科技。kkbida项目(凯京科技开源消息投递中间件)开发者,任职于凯京研发中心架构组,参与凯京科技统一支付平台、结算平台设计和开发,负责凯京科技开源中间件开发与维护。
欢迎加入凯京开源技术QQ群:613025121,和我们一起交流互联网应用的技术架构落地实践
凯京研发中心架构&运维部的工作主要分两大部分,架构组主要负责框架中间件的研究,如dubbo、apollo、skywalking、xxljob、分布式事务等、公司内开源项目以及公共服务公共组件的研发维护、新技术的引进以及落地等。运维组主要负责devops系统研发以及k8s容器环境的维护等工作。 架构组招聘目前架构组还有两个虚位以待,欢迎志同道合的你来和我们一起交流。简历可发送至邮箱: [email protected] |
今日新闻 |
点击排行 |
|
推荐新闻 |
图片新闻 |
|
专题文章 |
CopyRight 2018-2019 实验室设备网 版权所有 win10的实时保护怎么永久关闭 |