rabbitmq接口异常函数方法_RabbitMQ监控（三）：监控队列状态

#RabbitMQ 监控(三)

验证RabbitMQ健康运行只是确保消息通信架构可靠性的一部分，同时，你也需要确保消息通信结构配置没有遭受意外修改，从而避免应用消息丢失。

RabbitMQ Management HTTP API提供了一个方法允许你查看任何vhost上的任何队列：/api/queues//。你不仅可以查看配置详情，还可以查看队列的数据统计，例如队列消耗的内存，或者队列的平均消息吞吐量。使用curl测试一下该API，这里的/%2F还是代表默认的vhost(/)。

curl -u guest:guest http://127.0.0.1:15672/api/queues/%2F/springrabbitexercise

response

{

"consumer_details": [

{

"channel_details": {

"peer_host": "127.0.0.1",

"peer_port": 62679,

"connection_name": "127.0.0.1:62679 -> 127.0.0.1:5672",

"user": "guest",

"number": 2,

"node": "rabbit@localhost",

"name": "127.0.0.1:62679 -> 127.0.0.1:5672 (2)"

"arguments": [],

"prefetch_count": 1,

"ack_required": true,

"exclusive": false,

"consumer_tag": "amq.ctag-YImeU8Fm_VahDpxv8EAw2Q",

"queue": {

"vhost": "/",

"name": "springrabbitexercise"

}

"messages_details": {

"rate": 7357

"messages": 232517,

"messages_unacknowledged_details": {

"rate": 0.2

"messages_unacknowledged": 5,

"messages_ready_details": {

"rate": 7356.8

"messages_ready": 232512,

"reductions_details": {

"rate": 1861021.8

"reductions": 58754154,

...

"auto_delete": false,

"durable": true,

"vhost": "/",

"name": "springrabbitexercise",

"message_bytes_persistent": 2220250,

"message_bytes_ram": 2220250,

"message_bytes_unacknowledged": 40,

"message_bytes_ready": 2220210,

"message_bytes": 2220250,

"messages_persistent": 232517,

"messages_unacknowledged_ram": 5,

"messages_ready_ram": 232512,

"messages_ram": 232517,

"garbage_collection": {

"minor_gcs": 0,

"fullsweep_after": 65535,

"min_heap_size": 233,

"min_bin_vheap_size": 46422,

"max_heap_size": 0

"state": "running"

}

为了方便阅读，去掉了部分返回值，但是还是可以看到队列的很多信息。例如可以看到一个consumer的信息、消息占用的内存、队列的durable、auto_delete属性等。利用这些配置信息，新的健康监控程序可以通过API方法的输出来轻松监控队列的属性，并在发生变更时通知你。

就像之前编写健康检测程序那样，除了服务器、端口、vhost、用户名和密码之外，还需要知道：

* 队列的名称，以便监控其配置

* 该队列是否将durable和auto_delete选项打开

###清单3.1 检测队列配置

完整代码在我的github，下面代码中的@Data和@Slf4j都是插件lombok中的注解，想要了解的可自行百度。

1.定义查看队列信息的接口 RMQResource.java

@Path("api")

@Consumes({MediaType.APPLICATION_JSON})

@Produces({MediaType.APPLICATION_JSON})

public interface RMQResource {

/**

* Return a queue`s info

* @param vhost

* @param name

* @return {@link QueueInfo}

@GET

@Path("queues/{vhost}/{name}")

Response getQueueInfo(@PathParam("vhost") String vhost, @PathParam("name") String name);

}

2.定义查看队列接口的返回值 QueueInfo.java

@Data

public class QueueInfo {

private ConsumerDetails[] consumer_details;

/**

* unknown class

@JsonIgnore

private Object[] incoming;

/**

* unknown class

@JsonIgnore

private Object[] deliveries;

/**

* unknown class

@JsonIgnore

private Object arguments;

private Boolean exclusive;

//...

private Boolean auto_delete;

private Boolean durable;

private String vhost;

private String name;

/**

* unknown class

@JsonIgnore

private Object head_message_timestamp;

/**

* unknown class

@JsonIgnore

private Object recoverable_slaves;

private Long memory;

private Double consumer_utilisation;

private Integer consumers;

/**

* unknown class

@JsonIgnore

private Object exclusive_consumer_tag;

/**

* unknown class

@JsonIgnore

private Object policy;

@JsonFormat(pattern = "yyyy-MM-dd hh:mm:ss")

private Date idle_since;

}

3.检测队列配置 QueueConfigCheck.java

/**

* 检测队列配置

@Slf4j

public class QueueConfigCheck {

private final static RMQResource rmqResource = RMQApi.getService(RMQResource.class);

public static void checkQueueConfig(String vhost, CheckQueue queue) {

RMQConfig config = RMQConfig.Singleton.INSTANCE.getRmqConfig();

String host = config.getHost();

Response response = null;

try {

response = rmqResource.getQueueInfo(vhost, queue.getQueue_name());

} catch (Exception e) {

log.error("UNKNOWN: Could not connect to {}, cause {}", host, e.getMessage());

ExitUtil.exit(ExitType.UNKNOWN.getValue());

}

if (response == null || response.getStatus() == 404) {

log.error("CRITICAL: Queue {} does not exist.", queue.getQueue_name());

ExitUtil.exit(ExitType.CRITICAL.getValue());

} else if (response.getStatus() > 299) {

log.error("UNKNOWN: Unexpected API error : {}", response);

ExitUtil.exit(ExitType.UNKNOWN.getValue());

} else {

QueueInfo info = response.readEntity(QueueInfo.class);

if (!info.getAuto_delete().equals(queue.getAuto_delete())) {

log.warn("WARN: Queue {} - auto_delete flag is NOT {}", queue.getQueue_name(), info.getAuto_delete());

ExitUtil.exit(ExitType.WARN.getValue());

}

if (!info.getDurable().equals(queue.getDurable())) {

log.warn("WARN: Queue {} - durable flag is NOT {}", queue.getQueue_name(), info.getDurable());

ExitUtil.exit(ExitType.WARN.getValue());

}

log.info("OK: Queue {} configured correctly.", queue.getQueue_name());

ExitUtil.exit(ExitType.OK.getValue());

}

4.检测队列配置的方法参数 CheckQueue.java

@Data

public class CheckQueue {

private final String queue_name;

private final Boolean auto_delete;

private final Boolean durable;

public CheckQueue(String queue_name, Boolean auto_delete, Boolean durable) {

this.queue_name = queue_name;

this.auto_delete = auto_delete;

this.durable = durable;

}

5.运行检测程序

@Test

public void testQueueConfig() {

String queue_name = "springrabbitexercise";

Boolean auto_delete = false;

Boolean durable = true;

String vhost = "/";

CheckQueue queue = new CheckQueue(queue_name, auto_delete, durable);

QueueConfigCheck.checkQueueConfig(vhost, queue);

}

可以看到监控正常运行：

11:38:23.286 [main] INFO com.lanxiang.rabbitmqmonitor.check.QueueConfigCheck - OK: Queue springrabbitexercise configured correctly.

11:38:23.289 [main] INFO com.lanxiang.rabbitmqmonitor.terminate.ExitUtil - Status is OK

这段RabbitMQ队列检测的程序有一处修改，如果健康检测程序无法连接到API服务器的话，会返回EXIT_UNKNOWN。前一章的API ping健康检测要么成功要么失败，故障代码之间没有区别，但是队列检测API方法在失败时通过HTTP状态码提供了更多信息。如果HTTP状态码是404就代表尝试验证的队列不存在，检测失败并返回EXIT_CRITICAL。对于其他大于299的HTTP状态码，退出代码为EXIT_UNKNOWN。

在获取到RabbitMQ API的response之后，使用JSON进行解码，并且把得到的durable和auto_delete参数与期望的参数进行比较，如果参数和预期不相符的话，返回EXIT_WARNING或者EXIT_CRITICAL状态码。如果队列所有的配置都正确的话，那么就正确退出。

在了解我们对RabbitMQ做监控的原理之后，可以根据RabbitMQ Management HTTP API定制更多的监控，例如：

* /api/nodes，可以获取集群中每个节点的数据

* /api/queues//，可以获取队列的详细情况，例如消息处理的速率、积压的消息数量等。

除此之外还有许多其他API，我们要做的就是根据自身的业务逻辑和这些API来设计合理的监控脚本。RabbitMQ监控系列就到此结束啦，还是很可惜没有实战的机会吧，因为最近在工作变动期间，看了一下RabbitMQ实战这本书，兴起想写一下博客试试。

毕业快一年了，想养成写博客的习惯。正好最近也在工作变动中，能有闲暇时间尝试一下，博客写的比较水，多多包涵。

rabbitmq接口异常函数方法_RabbitMQ监控（三）：监控队列状态相关推荐

rabbitmq接口异常函数方法_分布式系统消息中间件——RabbitMQ的使用进阶篇
一 mandatory 参数上一篇文章中我们知道,生产者将消息发送到RabbitMQ的交换器中通过RoutingKey与BindingKey的匹配将之路由到具体的队列中以供消费者消费.那么当我们通过 ...
rabbitmq接口异常函数方法_[项目更新] 集成RabbitMQ队列与EventBus总线
(Blog.Core框架开发情况,着色部分为本次新增) 终于项目继续迭代更新了,在开源这两年多,也是感谢每一个支持Blog.Core项目的同学,同时也感谢每一个在生产环境中使用,并提出意见和建议的小伙 ...
JavaScript 用函数方法比较任意三个数字的最大值
JavaScript 用函数方法比较任意三个数字的最大值. 提示:这里只针对任意三个数字进行比较得最大值,使用 if 语句进行判断 . function getMax(num1, num2, num3 ...
阿里云短信接口，函数方法
调用方法:sendmsg(15112211211) /** * 方法功能:新版发送短信接口 * 开发时间:16-7-26 下午2:29 * $phone :需要发送目的地手机号 * $content; ...
OpenCV 3.1 imwrite()函数写入异常问题解决方法
OpenCV 3.1 imwrite()函数写入异常问题解决方法最近配置了OpenCV3.1版本,按照2.x的习惯写了一个保存图片的代码(测试证明该代码在2.4.11下运行正常),但是在使用imwr ...
C#（二）：数据类型、运算符、语句、类型转换、异常捕获、函数方法定义
数据类型.运算符.语句.类型转换.异常捕获.函数方法定义命名约定变量文本数字布尔值任意类型 `object` `dynamic` 声明局部变量获取类型的默认值数组数组定义方法 `x ...
mybatis接口中的方法重载_MyBatis的Mapper接口以及Example的实例函数及详解
一.mapper接口中的方法解析 mapper接口中的函数及方法方法功能说明 int countByExample(UserExample example) thorws SQLException ...
【Groovy】MOP 元对象协议与元编程 ( 使用 Groovy 元编程进行函数拦截 | 实现 GroovyInterceptable 接口 | 重写 invokeMethod 方法 )
文章目录一.GroovyInterceptable 接口简介二.重写 GroovyObject#invokeMethod 方法三.GroovyInterceptable 接口拦截效果四.完整代 ...
js中当等于最小值是让代码不执行_网页中JS函数自动执行常用三种方法
本文为大家分享了在网页中JS函数自动执行常用方法,供大家参考,具体内容如下一.JS方法 1.最简单的调用方式,直接写到html的body标签里面: 2.在JS语句调用: function myfun ...

rabbitmq接口异常函数方法_RabbitMQ监控（三）：监控队列状态

rabbitmq接口异常函数方法_RabbitMQ监控（三）：监控队列状态相关推荐

最新文章

热门文章