解决mqtt messageArrived()方法调用操作数据库方法断开连接问题

78 阅读 0 评论 52 点赞

我是靠谱客的博主潇洒烧鹅，最近开发中收集的这篇文章主要介绍解决mqtt messageArrived()方法调用操作数据库方法断开连接问题，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

在springboot中整合mqtt 在回调方法 messageArrived()获取数据如果直接操作数据库会使用连接断开

解决方法：把要操作数据库的方法封装到回调类，在本类中调用

先把mapper注入进来

具体原因可以看我上一篇博客 https://blog.csdn.net/qq_41249513/article/details/108464133

整个类写法

package com.laima.operate.voltoperate.mqtt;

/**
 * @Author: wrc
 * @Classname PushCallback
 * @Description TODO
 * @Date 2020/9/7 16:52
 * @Created wrc
 */

import com.laima.operate.voltoperate.dao.ElecMeasuringPointDao;
import com.laima.operate.voltoperate.entity.ElecMeasuringPointEntity;
import com.laima.operate.voltoperate.service.ElecMeasuringPointService;
import com.sun.javafx.image.impl.IntArgb;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * 发布消息的回调类
 *
 * 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。
 * 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。
 * 在回调中，将它用来标识已经启动了该回调的哪个实例。
 * 必须在回调类中实现三个方法：
 *
 *  public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
 *
 *  public void connectionLost(Throwable cause)在断开连接时调用。
 *
 *  public void deliveryComplete(MqttDeliveryToken token))
 *  接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。
 *  由 MqttClient.connect 激活此回调。
 *
 */
@Component
public class PushCallback implements MqttCallback {

    private static final Logger LOGGER = LoggerFactory.getLogger(PushCallback.class);


    @Autowired
    public ElecMeasuringPointDao elecMeasuringPointDao;

    public  static PushCallback PushCallback ;

    @PostConstruct //通过@PostConstruct实现初始化bean之前进行的操作
    public void init() {
        System.out.println("初始化");
        PushCallback = this;
        PushCallback.elecMeasuringPointDao = this.elecMeasuringPointDao;
    }

    /**
     * 主题
     */
    private String topic = "voltceshi";//测试

//    private String topic = "正式环境主题";//正式

    private MQTTSubsribe mqttSubsribe;
//    private MQTTSubsribe mqttSubsribe = new MQTTSubsribe();



    public PushCallback(MQTTSubsribe subsribe) {
        this.mqttSubsribe = subsribe;
    }

    //这是封装mapper的插入方法，解决插入数据连接断开
    public int insert(ElecMeasuringPointEntity elecMeasuringPointEntity ){
        return elecMeasuringPointDao.insert(elecMeasuringPointEntity);
    }


    @Override
    public void connectionLost(Throwable cause) {
        // 连接丢失后，一般在这里面进行重连
        LOGGER.info("---------------------连接断开，可以做重连");
        // deliveryComplete(null);

        while (true){
            try {//如果没有发生异常说明连接成功，如果发生异常，则死循环
                Thread.sleep(1000);
                mqttSubsribe.init();
                break;
            }catch (Exception e){
//                e.printStackTrace();
                continue;
            }
        }

    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("deliveryComplete---------" + token.isComplete());
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // subscribe后得到的消息会执行到这里面
        String result = new String(message.getPayload(),"UTF-8");
        System.out.println("接收消息主题 : " + topic);
        System.out.println("接收消息Qos : " + message.getQos());
        System.out.println("接收消息内容 : " + result);

        //这里可以针对收到的消息做处理
        ElecMeasuringPointEntity elecMeasuringPointEntity = new ElecMeasuringPointEntity();
        elecMeasuringPointEntity.setUEnterpriseId(1111);
        //插入方法
        PushCallback.elecMeasuringPointDao.insert(elecMeasuringPointEntity);

        //elecMeasuringPointService.insert(elecMeasuringPointEntity);
    }



}