提交粮情测试协议和流程

jiazx0107@163.com

2023-11-08 768308ab47041062d9e97745099784acf297286b

提交粮情测试协议和流程

 src/main/java/com/fzzy/api/data/GatewayDeviceProtocol.java

@@ -18,7 +18,7 @@
    DEVICE_IDCARD_HTTP("DEVICE_IDCARD_HTTP", "身份证-HTTP协议"),
    DEVICE_LED_HTTP("DEVICE_LED_HTTP", "LED-HTTP协议"),
    DEVICE_LPR_SDK_HK("DEVICE_LPR_SDK_HK", "车牌识别-SDK海康"),
    DEVICE_TEST("DEVICE_TEST", "空协议");
    DEVICE_TEST("DEVICE_TEST", "测试协议");


    private String code;

 src/main/java/com/fzzy/api/utils/NumberUtil.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,49 @@
package com.fzzy.api.utils;
 
import java.math.BigDecimal;
import java.text.DecimalFormat;

/**
 * æ•°å­—格式化工具类
 */
public class NumberUtil {

    /**
     * å¯¹double类型的数值保留指定位数的小数。<br>
     * è¯¥æ–¹æ³•èˆå…¥æ¨¡å¼ï¼šå‘“最接近的”数字舍入，如果与两个相邻数字的距离相等，则为向上舍入的舍入模式。<br>
     * <b>注意：</b>如果精度要求比较精确请使用 keepPrecision(String number, int precision)方法
     * @param number  è¦ä¿ç•™å°æ•°çš„æ•°å­—
     * @param precision å°æ•°ä½æ•°
     * @return double å¦‚果数值较大，则使用科学计数法表示
     */
    public static double keepPrecision(Double number, int precision) {
        if(null == number || 0.0 == number ) return 0.0;
        BigDecimal bg = new BigDecimal(number);
        return bg.setScale(precision, BigDecimal.ROUND_HALF_UP).doubleValue();
    }
 
    /**
     * å¯¹float类型的数值保留指定位数的小数。<br>
     * è¯¥æ–¹æ³•èˆå…¥æ¨¡å¼ï¼šå‘“最接近的”数字舍入，如果与两个相邻数字的距离相等，则为向上舍入的舍入模式。<br>
     * <b>注意：</b>如果精度要求比较精确请使用 keepPrecision(String number, int precision)方法
     * @param number  è¦ä¿ç•™å°æ•°çš„æ•°å­—
     * @param precision å°æ•°ä½æ•°
     * @return float å¦‚果数值较大，则使用科学计数法表示
     */
    public static float keepPrecision(Float number, int precision) {
        if(null == number) return 0f;
        BigDecimal bg = new BigDecimal(number);
        return bg.setScale(precision, BigDecimal.ROUND_HALF_UP).floatValue();
    }
    /**
     * double转字符串，避免出现科学计数法
     * @param d
     * @return
     */
    public static String doubleToStr(Double d) {
        if(null == d) return "";
        DecimalFormat df = new DecimalFormat("0.0");
        return df.format(d);
    }
 
}

 src/main/java/com/fzzy/gateway/GatewayRunner.java

@@ -22,11 +22,20 @@
    private ApiInitService apiInitService;
    @Autowired
    private MqttPublishService mqttPublishService;

    @Autowired
    private  GatewayTimerScheduled scheduled;

    @Override
    public void run(String... args) throws Exception {
        mqttPublishService.init();
        //执行初始化方案
        apiInitService.init();


        //获取气象信息
        scheduled.doWeatherExe();

    }

}

 src/main/java/com/fzzy/gateway/GatewayTimerScheduled.java

@@ -1,14 +1,18 @@
package com.fzzy.gateway;

import com.alibaba.fastjson.JSON;
import com.fzzy.api.data.ApiParam;
import com.fzzy.api.entity.ApiConfs;
import com.fzzy.api.service.*;
import com.fzzy.api.utils.ContextUtil;
import com.fzzy.api.utils.RedisUtil;
import com.fzzy.gateway.api.GatewayRemoteManager;
import com.fzzy.gateway.data.WeatherWebDto;
import com.fzzy.gateway.entity.GatewayConf;
import com.fzzy.gateway.service.GatewayConfService;
import com.fzzy.gateway.util.GatewayHttpUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
@@ -24,6 +28,12 @@
@Slf4j
@Component(GatewayTimerScheduled.BEAN_ID)
public class GatewayTimerScheduled {

    public static String DEFAULT_URL = "https://v0.yiketianqi.com/api?unescape=1&version=v61&appid={appId}&appsecret={appsecret}&cityid={cityid}";
    public static String DEFAULT_APP_ID = "49421971";
    public static String DEFAULT_APP_SECRET = "JmJE48Fv";
    public static String DEFAULT_CITYID = "101270101";//成都


    public static final String BEAN_ID = "gateway.timerScheduled";

@@ -46,6 +56,44 @@

    }

    /**
     * æ¯é—´éš”30分钟执行一次
     */
    @Scheduled(cron = "0 0/30 * * * ?")
    public void scheduled30() {

        //定时获取气象信息
        doWeatherExe();
    }

    public void doWeatherExe() {

        try {
            String url = DEFAULT_URL;
            url = url.replace("{appId}", DEFAULT_APP_ID).replace("{appsecret}", DEFAULT_APP_SECRET).replace("{cityid}", DEFAULT_CITYID);

            String result = GatewayHttpUtil.doGet(url, null);

            if (null == result) {
                log.error("当前外网获取气象信息失败……");
                return;
            }
            WeatherWebDto dto = JSON.parseObject(result, WeatherWebDto.class);
            if (StringUtils.isNotEmpty(dto.getErrcode())) {
                log.error("当前外网获取气象信息异常:{}", dto.getErrmsg());
                return;
            }

            WeatherWebDto.contextMap.put("default", dto);


            log.info("===========================系统定时获获取气象信息===={}==================",dto);

        } catch (Exception e) {

        }
    }


    /**
     * æ‰§è¡Œç½‘关心跳
@@ -63,5 +111,5 @@
            gatewayRemoteManager.getRemoteService(conf.getPushProtocol()).heartbeat(conf);
        }
    }
    

}

 src/main/java/com/fzzy/gateway/api/GatewaySyncGranService.java

@@ -1,6 +1,7 @@
package com.fzzy.gateway.api;


import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.hx2023.data.KafaGrainData;
import com.fzzy.gateway.hx2023.data.*;

@@ -25,4 +26,11 @@
    public KafaGrainData syncGrain(SyncReqData reqData);


    /**
     * åŒæ­¥ç²®æƒ…信息返回JSON报文
     *
     * @param reqData
     * @return
     */
    public BaseResp syncGrain2(SyncReqData reqData);
}

 src/main/java/com/fzzy/gateway/data/BaseResp.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,14 @@
package com.fzzy.gateway.data;

import lombok.Data;

@Data
public class BaseResp {

    private int code = 200;

    private String msg = "成功";


    private String data;
}

 src/main/java/com/fzzy/gateway/data/WeatherWebDto.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,51 @@
package com.fzzy.gateway.data;

import lombok.Data;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

/**
 * 
 * @author: andy.jia
 * @description: å¤–网气象解析使用DTO
 * @version:
 * @data:2019å¹´12月7日
 *
 */
@Data
public class WeatherWebDto implements Serializable {


    public static Map<String,WeatherWebDto> contextMap = new HashMap<>();

    /**
     * 
     */
    private static final long serialVersionUID = 1L;
	
	
    private String errcode;//错误编码
    private String errmsg;//错误信息
	
    private String cityid;// å½“前城市ID
    private String city;//城市名称
    private String update_time;// æ°”象台更新时间
    private String date;// æ—¥æœŸ
    private String week;// æ˜ŸæœŸ
    private String wea;// å¤©æ°”情况
    private String wea_img;// å¤©æ°”对应图标(xue, lei, shachen, wu, bingbao, yun, yu,yin, qing)
    private String tem;// å½“前温度
    private String air;// ç©ºæ°”质量
    private String air_pm25;// PM2.5
    private String air_level;// ç©ºæ°”质量等级
    private String air_tips;// ç©ºæ°”质量描述
    private String humidity;// æ¹¿åº¦
    private String visibility;// èƒ½è§åº¦
    private String pressure;// æ°”压hPa
    private String win;// é£Žå‘
    private String win_speed;// é£Žé€Ÿç­‰çº§
    private String win_meter;// é£Žé€Ÿ å¦‚: 12km/h
    private String alarm;// é¢„警信息
}

 src/main/java/com/fzzy/gateway/entity/GatewayDevice.java

@@ -105,4 +105,13 @@
    @PropertyDef(label = "备注", description = "备注信息")
    private String remark;


    @Column(name = "CABLE_RULE_", length = 20)
    @PropertyDef(label = "布线规则", description = "平方仓表示层行列，筒仓表示每圈的列数")
    private String cableRule;

    @Column(name = "CABLE_CIR_", length = 20)
    @PropertyDef(label = "筒仓层规则", description = "针对筒仓")
    private String cableCir;

}

 src/main/java/com/fzzy/gateway/hx2023/ScConstant.java

@@ -12,6 +12,13 @@

    public static String MESSAGE_TYPE_REPORT_PROPERTY = "REPORT_PROPERTY";

    public static String MESSAGE_TYPE_INVOKE_FUNCTION = "INVOKE_FUNCTION";

    /**
     * ç²®æƒ…采指令
     */
    public static String FUNCTION_getTAndRHInfo = "getTAndRHInfo";


    public static String getMessageId() {
        return System.currentTimeMillis() + RandomUtils.nextInt(1000) + "";

 src/main/java/com/fzzy/gateway/hx2023/data/ClientHeaders.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,12 @@
package com.fzzy.gateway.hx2023.data;

import lombok.Data;

@Data
public class ClientHeaders {

    private String productId;
    private String deviceName;
    private String orgId;
    private String msgId;
}

 src/main/java/com/fzzy/gateway/hx2023/data/CloudHeaders.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,11 @@
package com.fzzy.gateway.hx2023.data;

import lombok.Data;

@Data
public class CloudHeaders {

    private String productId;
    private String deviceName;
    private String orgId;
}

 src/main/java/com/fzzy/gateway/hx2023/data/CloudSendData.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,28 @@
package com.fzzy.gateway.hx2023.data;

import lombok.Data;

import java.util.List;

/**
 * äº‘端发动的报文
 */
@Data
public class CloudSendData {

    private CloudHeaders headers;

    private String functionId;

    private String messageType;

    private String messageId;

    private String deviceId;

    private String timestamp;

    private boolean success;

    private List<InputData> inputs;
}

 src/main/java/com/fzzy/gateway/hx2023/data/GrainData.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,29 @@
package com.fzzy.gateway.hx2023.data;

import lombok.Data;


/**
 * ç²®æƒ…信息
 */
@Data
public class GrainData {

    //设备编码
    private String deviceId;
    private ClientHeaders headers;

    //消息 ID
    private String messageId;

    private String messageType = "INVOKE_FUNCTION_REPLY";

    private boolean success = true;

    private String timestamp;

    private GrainOutPut outPut;

    private GrainWeather weatherStation;

}

 src/main/java/com/fzzy/gateway/hx2023/data/GrainOutPut.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,42 @@
package com.fzzy.gateway.hx2023.data;

import com.alibaba.fastjson2.JSONObject;
import lombok.Data;

import java.util.List;


/**
 * ç²®æƒ…信息
 */
@Data
public class GrainOutPut {

    //模块识别码-常量，固定传 apiTemperature
    private String apISource = "apiTemperature";

    //整仓平均温度
    private String avgTemperature;

    //整仓最高温度
    private String maxTemperature;

    //整仓最低温度
    private String minTemperature;

    private String minX = "0";

    private String minY = "0";

    private String minZ = "0";

    private String maxX = "0";

    private String maxY = "0";

    private String maxZ = "0";


    private List<GrainTemp> temperature;

}

 src/main/java/com/fzzy/gateway/hx2023/data/GrainTemp.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,45 @@
package com.fzzy.gateway.hx2023.data;

import lombok.Data;

/**
 * ç²®æƒ…信息详细
 */
@Data
public class GrainTemp {

    //根号
    private String cableNum;

    //层号
    private String layerNumber;

    //温度值
    private String temperature;

    //索引，从0开始
    private String position;

//    //该温度点所在的列,平方仓、地下仓必填
//    private String linex;
//
//    //该温度点所在的行,平方仓、地下仓必填
//    private String rowy;

    //浅圆仓、筒仓必填，示例：{\"totalCircle\":3,\"smallCircle\":\"4,10,16\"}，totalCircle：总圈数，smallCircle：每圈有几根缆
    private String total_circle;


    //具体圈数--浅圆仓、筒仓必填
    private String circle;

    public GrainTemp() {
    }

    public GrainTemp(String cableNum, String layerNumber, String temperature, String position) {
        this.cableNum = cableNum;
        this.layerNumber = layerNumber;
        this.temperature = temperature;
        this.position = position;
    }
}

 src/main/java/com/fzzy/gateway/hx2023/data/GrainWeather.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,31 @@
package com.fzzy.gateway.hx2023.data;

import lombok.Data;

@Data
public class GrainWeather {

    private String airPressure;

    private String humidity;

    private String id;

    private String messageId;

    private String pm;

    private String radiation;

    private String rainfallAmount;

    private String temperature;

    private String windAngle;

    private String windDirection;

    private String windPower;

    private String windSpeed;
}

 src/main/java/com/fzzy/gateway/hx2023/data/InputData.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,10 @@
package com.fzzy.gateway.hx2023.data;

import lombok.Data;

@Data
public class InputData {

    private String name;
    private String value;
}

 src/main/java/com/fzzy/gateway/hx2023/data/SyncReqData.java

@@ -1,5 +1,6 @@
package com.fzzy.gateway.hx2023.data;

import com.fzzy.gateway.entity.GatewayDevice;
import lombok.Data;

@Data
@@ -16,4 +17,8 @@
    private String functionId;

    private String jsonData;

    private GatewayDevice device;

    private boolean autoReplay;
}

 src/main/java/com/fzzy/gateway/hx2023/service/HxGatewaySyncGrainImpl.java

@@ -1,11 +1,21 @@
package com.fzzy.gateway.hx2023.service;

import com.alibaba.fastjson2.JSONObject;
import com.fzzy.api.data.GatewayDeviceProtocol;
import com.fzzy.api.utils.NumberUtil;
import com.fzzy.gateway.api.GatewaySyncGranService;
import com.fzzy.gateway.hx2023.data.KafaGrainData;
import com.fzzy.gateway.hx2023.data.SyncReqData;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.data.WeatherWebDto;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.ScConstant;
import com.fzzy.gateway.hx2023.data.*;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * ç½‘关与粮情分机通讯和解析
@@ -18,11 +28,105 @@

    @Override
    public String getGrainProtocol() {
        return null;
        return GatewayDeviceProtocol.DEVICE_TEST.getCode();
    }

    @Override
    public KafaGrainData syncGrain(SyncReqData reqData) {


        return null;
    }


    @Override
    public BaseResp syncGrain2(SyncReqData reqData) {

        GatewayDevice device = reqData.getDevice();


        if (StringUtils.isEmpty(device.getCableCir())) {
            return getGrainTest1(reqData, device);
        }


        BaseResp resp = new BaseResp();
        resp.setCode(500);
        resp.setMsg("没有匹配到规则");
        return resp;
    }

    private BaseResp getGrainTest1(SyncReqData reqData, GatewayDevice device) {

        String[] cableRule = device.getCableRule().split("-");

        int cableZ = Integer.valueOf(cableRule[0]);
        int cableY = Integer.valueOf(cableRule[1]);
        int cableX = Integer.valueOf(cableRule[2]);
        int sumNum = cableZ * cableY * cableX;


        WeatherWebDto weather = WeatherWebDto.contextMap.get("default");
        double tMIn = 20, tMax = 25;
        if (null != weather) {
            double tOut = Double.valueOf(weather.getTem());
            tMIn = tOut - 2;
            tMax = tOut + 3;
        }


        //数据封装
        GrainData grain = new GrainData();
        grain.setMessageId(ScConstant.getMessageId());
        grain.setDeviceId(device.getDeviceId());
        grain.setTimestamp(System.currentTimeMillis() + "");

        ClientHeaders headers = new ClientHeaders();
        headers.setDeviceName(device.getDeviceName());
        headers.setProductId(device.getProductId());
        headers.setOrgId(device.getOrgId());
        headers.setMsgId(ScConstant.getMessageId());
        grain.setHeaders(headers);

        GrainOutPut outPut = new GrainOutPut();

        outPut.setAvgTemperature(NumberUtil.keepPrecision((tMax + tMIn) / 2, 1) + "");
        outPut.setMinTemperature(tMax + "");
        outPut.setMaxTemperature(tMIn + "");


        List<GrainTemp> temperature = new ArrayList<>();
        //根号
        int cableNum = 1, position = 0;

        double curTemp = tMIn;
        double randomNumber = tMIn;
        int x = 0, y = 0, z = 0;
        for (int i = 0; i < sumNum; i++) {
            randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;
            curTemp = NumberUtil.keepPrecision(randomNumber, 1);
            position = i;
            z = i % cableZ + 1;
            x = i / (cableZ * cableY);
            y = x * (cableZ * cableY);
            y = (i - y) / cableZ;

            //根号
            cableNum = (i / cableZ) + 1;


            temperature.add(new GrainTemp(cableNum + "", z + "", curTemp + "", position + ""));
        }

        outPut.setTemperature(temperature);

        grain.setOutPut(outPut);


        BaseResp resp = new BaseResp();
        resp.setData(JSONObject.toJSONString(grain));

        return resp;
    }
}


 src/main/java/com/fzzy/gateway/hx2023/service/OnReceiveMqttService.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,82 @@
package com.fzzy.gateway.hx2023.service;

import com.alibaba.fastjson2.JSONObject;
import com.fzzy.gateway.GatewayUtils;
import com.fzzy.gateway.api.GatewayRemoteManager;
import com.fzzy.gateway.hx2023.data.CloudSendData;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.ScConstant;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.hx2023.data.SyncReqData;
import com.fzzy.mqtt.MqttProviderConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * æ ¹æ®æŽ¥å—到的MQTT信息执行
 */
@Slf4j
@Component
public class OnReceiveMqttService {

    @Resource
    private GatewayRemoteManager gatewayRemoteManager;
    @Resource
    private MqttProviderConfig providerClient;


    /**
     * å½“前接收到云端发送信息
     *
     * @param message
     */
    public void onReceiveMessage(String message) {

        try {
            CloudSendData cloudSendData = JSONObject.parseObject(message, CloudSendData.class);


            String functionId = cloudSendData.getFunctionId();

            //粮情采集
            if (ScConstant.FUNCTION_getTAndRHInfo.equals(functionId)) {
                getTAndRHInfo(cloudSendData);
            }

        } catch (Exception e) {

        }

    }

    private void getTAndRHInfo(CloudSendData cloudSendData) {

        String deviceId = cloudSendData.getDeviceId();

        GatewayDevice device = GatewayUtils.getCacheByDeviceId(deviceId);

        SyncReqData syncReqData = new SyncReqData();
        syncReqData.setDeviceId(deviceId);
        syncReqData.setMessageId(cloudSendData.getMessageId());
        syncReqData.setMessageType(cloudSendData.getMessageType());
        syncReqData.setFunctionId(cloudSendData.getFunctionId());
        syncReqData.setAutoReplay(true);
        syncReqData.setDevice(device);

        BaseResp resp = gatewayRemoteManager.getSyncGrainService(device.getSyncProtocol()).syncGrain2(syncReqData);

        //自动推送
        if (200 == resp.getCode() && syncReqData.isAutoReplay()) {
            String topic = "/${productId}/${deviceId}/properties/report";

            topic = topic.replace("${productId}", device.getProductId()).replace("${deviceId}", device.getDeviceId());

            providerClient.publish(topic, resp.getData());

            log.info("=======粮情推送==========={}", resp.getData());
        }
    }
}

 src/main/java/com/fzzy/gateway/service/GatewayDeviceService.java

@@ -113,6 +113,9 @@






    /**
     * gatewayDeviceService#ajaxTestWeight
     * åœ°ç£…推送测试

 src/main/java/com/fzzy/gateway/view/GatewayDevice.view.xml

@@ -111,8 +111,16 @@
        <Property name="label">自定义仓库编码</Property>
      </PropertyDef>
      <PropertyDef name="productId">
        <Property/>
        <Property></Property>
        <Property name="label">设备类型KEY</Property>
      </PropertyDef>
      <PropertyDef name="cableRule">
        <Property/>
        <Property name="label">布线规则</Property>
      </PropertyDef>
      <PropertyDef name="cableCir">
        <Property/>
        <Property name="label">筒仓层规则</Property>
      </PropertyDef>
    </DataType>
  </Model>
@@ -332,6 +340,16 @@
              <Property name="editorType">TextArea</Property>
              <Editor/>
            </AutoFormElement>
            <AutoFormElement>
              <Property name="name">cableRule</Property>
              <Property name="property">cableRule</Property>
              <Editor/>
            </AutoFormElement>
            <AutoFormElement>
              <Property name="name">cableCir</Property>
              <Property name="property">cableCir</Property>
              <Editor/>
            </AutoFormElement>
          </AutoForm>
        </Container>
      </Children>

 src/main/java/com/fzzy/mqtt/MqttConsumerCallBack.java

@@ -1,17 +1,27 @@
package com.fzzy.mqtt;

import com.fzzy.gateway.hx2023.service.OnReceiveMqttService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class MqttConsumerCallBack implements MqttCallback {


    @Autowired
    private OnReceiveMqttService onReceiveMqttService;

    /**
     * å®¢æˆ·ç«¯æ–­å¼€è¿žæŽ¥çš„回调
     */
    @Override
    public void connectionLost(Throwable throwable) {
        System.out.println("与服务器断开连接，可重连");
        log.info("与服务器断开连接，可重连");
    }

    /**
@@ -20,10 +30,17 @@
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {

        System.out.println(String.format("接收消息主题 : %s", topic));
        System.out.println(String.format("接收消息Qos : %d", message.getQos()));
        System.out.println(String.format("接收消息内容 : %s", new String(message.getPayload())));
        System.out.println(String.format("接收消息retained : %b", message.isRetained()));

        String messageStr = new String(message.getPayload());


        log.info(String.format("接收消息主题 : %s", topic));
        log.info(String.format("接收消息Qos : %d", message.getQos()));
        log.info(String.format("接收消息内容 : %s", messageStr));

        log.info(String.format("接收消息retained : %b", message.isRetained()));

        onReceiveMqttService.onReceiveMessage(messageStr);
    }

    /**
@@ -31,6 +48,6 @@
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        System.out.println(String.format("接收消息成功"));
        log.info(String.format("接收消息成功"));
    }
}

 src/main/java/com/fzzy/mqtt/MqttConsumerConfig.java

@@ -14,6 +14,9 @@

    @Autowired
    private MqttProperties mqttProperties;
    @Autowired
    private MqttConsumerCallBack mqttConsumerCallBack;

    /**
     * å®¢æˆ·ç«¯å¯¹è±¡
     */
@@ -50,7 +53,8 @@
            //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
            options.setWill("willTopic", (mqttProperties.getClientInId() + "与服务器断开连接").getBytes(), 0, false);
            //设置回调
            client.setCallback(new MqttConsumerCallBack());
            // client.setCallback(new MqttConsumerCallBack());
            client.setCallback(mqttConsumerCallBack);
            client.connect(options);
            //订阅主题
            //消息等级，和主题数组一一对应，服务端将按照指定等级给订阅了主题的客户端推送消息