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

news2024/9/24 19:04:48

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));
        //正确的basic
        String 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 {
    @Autowired
    private RabbitMQHttpService rabbitMQHttpService;
    @Test
    public void test(){
        rabbitMQHttpService.getQueueMessageCount("guest","guest","ordering_ok_wms");
    }
    @Test
    public 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();
    }