Merge remote-tracking branch 'origin/xm-20221107' into xm-20221107

liuhaonan

2022-11-18 b531004a3cbd33d7bb9f5b7dce06ddd4f5e7ec9a

 ximon-admin/src/main/java/com/sandu/ximon/admin/localMQTT/util/MqttClientUtil.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,92 @@
package com.sandu.ximon.admin.localMQTT.util;

import com.alibaba.fastjson.JSON;
import com.sandu.ximon.admin.localMQTT.client.MqttClientManager;
import com.sandu.ximon.admin.localMQTT.model.LocalMqttMsg;
import com.sandu.ximon.admin.utils.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static java.lang.Thread.sleep;

/**
 * @author van
 * @version 1.0
 * msg：MQTT客户端工具类
 * @date 2022/11/9 16:32
 */
@Slf4j
public class MqttClientUtil {

    private static MqttClientUtil mqttClientUtil = new MqttClientUtil();

    public static MqttClientUtil getInstance() {
        return mqttClientUtil;
    }

    public static String publishPrefix = "v1/devices/request/";


    public static String clientId = "java_server_msg";

    public static final Map<String, String> MQTT_RETURN_FRAME_MAP = new ConcurrentHashMap<>();

    public static String sendMqttMsg(String topic, String content) {
        try {
            LocalMqttMsg localMqttMsg = new LocalMqttMsg();
            localMqttMsg.setPayload(content);

            MqttMessage message = new MqttMessage(JSON.toJSONString(localMqttMsg).getBytes());

             message.setQos(0);
            MqttClient mqttClient = MqttClientManager.getInstance().getMqttClientById(clientId);

            if (null == mqttClient) {
                log.error("Not exist mqttClient where it's clientId is {}", clientId);
                return topic;
            }

            long start = System.currentTimeMillis();
            mqttClient.publish(publishPrefix+topic, message);
            log.info("{}",publishPrefix+topic);
            log.info("{}", message);

            /**
             * å®žçŽ°ä¼ªåŒæ­¥
             * */
            try {
                String returnFrame = null ;
                returnFrame = MQTT_RETURN_FRAME_MAP.get(topic);

                for (int i = 0;i < 50;i++){
                    if (StringUtil.strIsNullOrEmpty(returnFrame)){
                        sleep(100);
                        returnFrame = MQTT_RETURN_FRAME_MAP.get(topic);
                    }else {
                        log.info("返回时间：{} ms",System.currentTimeMillis() - start) ;
                        String remove = MQTT_RETURN_FRAME_MAP.remove(topic);
                        log.info("remove结果：{} ",remove);
                        return returnFrame;
                    }
                }


            } catch (InterruptedException e) {

            }

        } catch (MqttException e) {
            log.error("MqttClient send msg faild!", e);
            return("通信超时");
        }
        return null;
    }



}