XXL-Job 监控消息队列消息数量预警

1、什么是Basic Authentication认证

Basic Authentication 是一种常用的 HTTP 认证机制，用于保护 Web 资源免受未授权访问。在这种认证方式中，客户端（通常是浏览器）需要在 HTTP 请求头中提供用户凭据（通常是用户名和密码），这些凭据以 Base64 编码的字符串形式包含在 "Authorization" 头字段中

当用户访问被保护起来的资源时，服务器会判断Header里面有没有Authorization字段，如果没有会返回status = 401状态码。

在请求Authenticate头部指定类型为Basic，客户端随后会提提示用户输入用户名和密码，并把它们以username:password的形式编码为Base64字符串，在后续的请求中通过Authorization头发送给服务器

没有请求头的情况下，直接使用RabbitMQAPI查看所有的队列，浏览器会弹出提示框

https://www.jb51.net/program/3242837xn.htm

2、在Java中使用HttpRequest.get请求时如何通过Basic Authentication认证

Basic Authentication认证会将用户输入的账号和密码拼接成 username:password后转化成base64格式,拼接上Basic

拼接效果如下所示

"Basic Z3Vlc3Q6Z3Vlc3Q= "

Java如何实现拼接并发送请求

指定查看某个队列和所有队列的消息数量

public void getQueueMessageCount(String username,String password,String queueName) {	//获得传递过来的账号密码	//手动拼接成指定格式String basic = username+":"+password;			//将拼接后的账号密码编码成base64格式	String encode = Base64.encode(basic.getBytes(StandardCharsets.UTF_8));	//将编译好的账号密码拼接上Basic	//真正的Basic	String Basic = "Basic"+ " "+ encode;	//请求路径	String url = "http://192.168.29.131:15672/api/queues/"+"%2f"+"/"+queueName;	//通过HttpRequest发送get请求     HttpResponse authorization = HttpRequest.get(url).header("Authorization", Basic).execute();//访问成功if(authorization.getStatus() == 200){String body = authorization.body();//获得所有的信息HashMap bodyList = JSONUtil.toBean(body, HashMap.class);Integer messages = (Integer)bodyList.get("messages");//判断消息数量是否足够if (messages > 3){//钉钉报警SendMessage(queueName);return;}else {System.out.println("消息队列信息正常，无事发生");}}
}//查看所有队列的消息 public void getQueueMessageCount(String username,String password) {//获得传递过来的账号密码String basic = username+":"+password;//base64编码/*不指定UTF-8，默认会使用系统编码格式，但是在不同的操作系统中大概率会出错，建议明确编码格式*/String encode = Base64.encode(basic.getBytes(StandardCharsets.UTF_8));//正确的basicString Basic = "Basic" +" "+encode;String url = "http://192.168.29.131:15672/api/queues/";HttpResponse authorization = HttpRequest.get(url).header("Authorization", Basic).execute();//访问成功if(authorization.getStatus() == 200){String body = authorization.body();//获得所有的信息List<Map> bodyList = JSONUtil.toList(body, Map.class);for (Map map : bodyList) {Integer message = (Integer) map.get("messages");//判断消息数量是否足够if (message > 3 ){//钉钉报警SendMessage(map.get("name").toString());}else {log.debug("{}消息总量并没有超过限定值，无事发生",map.get("name"));}}}String mm = "";}/*钉钉发送消息*/ private void SendMessage(String queueName ){String url = "https://oapi.dingtalk.com/robot/send?access_token=8fbcdf4c54bd3954b1f71993e6eeac1ea78f126d669f028d01349df44b37ead9";//存放at内部信息JSONArray array = new JSONArray();array.add("17839697920");//存放机器人的一些Json字符串JSONObject msg = new JSONObject();msg.set("msgtype","text");//String nickName = null;msg.set("text",new JSONObject().set("content","警告,消息队列： "+queueName+" 中的消息已经超出设定值，请尽快进行处理" ));msg.set("at",new JSONObject().set("atMobiles",array));String json = JSONUtil.toJsonStr(msg);HttpRequest.post(url).body(json).execute().body();}

测试方法

@SpringBootTest
public class MQMessageCount {@Autowiredprivate RabbitMQHttpService rabbitMQHttpService;@Testpublic void test(){rabbitMQHttpService.getQueueMessageCount("guest","guest","ordering_ok_wms");}@Testpublic void test2(){rabbitMQHttpService.getQueueMessageCount("guest","guest");}
}

使用XXl_Job监控队列消消息是否超出指定数量

XXL_Job调度配置

{

"url":"http://192.169.31.400:15672/api/queues/%2f/canal_q01","method":"GET","body":"","header":"Authorization:Basic Z3Vlc3Q6Z3Vlc3Q= ","count":"1"

}

@XxlJob("httpJobHandler")public void httpJobHandler() throws Exception {String param = XxlJobHelper.getJobParam();HashMap map = JSONUtil.toBean(param, HashMap.class);String url = map.get("url").toString();String method = map.get("method").toString();String body = map.get("body").toString();Integer count = Integer.valueOf(map.get("count").toString());String[] headers = map.get("header").toString().split(":");String ResultBody = HttpRequest.of(url).method(Method.GET).header(headers[0], headers[1]).execute().body();//通过ResultBody获得返回过来的数据并转换成json字符串Map ResultMap = JSONUtil.toBean(ResultBody, HashMap.class);Integer messagesCount = (Integer)ResultMap.get("messages");//获得队列名称String queueName = (String)ResultMap.get("name");if (messagesCount > count){SendMessage(queueName);}}private void SendMessage(String queueName ){String url="https://钉钉Api";//存放at内部信息JSONArray array = new JSONArray();array.add("17839697920");//存放机器人的一些Json字符串JSONObject msg = new JSONObject();msg.set("msgtype","text");//String nickName = null;msg.set("text",new JSONObject().set("content","警告,消息队列： "+queueName+" 中的消息已经超出设定值，请尽快进行处理" ));msg.set("at",new JSONObject().set("atMobiles",array));String json = JSONUtil.toJsonStr(msg);HttpRequest.post(url).body(json).execute().body();}