SpringBoot集成MQTT配置-摩杜云开发者社区

springboot 集成mqtt实现发布、订阅消息的功能

首先在pom.xml文件中加入依赖

<dependency>
	<groupId>org.springframework.integration</groupId>
	<artifactId>spring-integration-mqtt</artifactId>
</dependency>

application.yml配置

mqtt:
    host: tcp://xx.xx.xx.xx:1883
    topics: scrm_user/#
    clientId: clientId
    maxInflight: 20
    username: admin
    password: public

MqttFactory

import com.alibaba.fastjson.JSONObject;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

/**
 * @Function: MQTTConfig
 * @Description: MQTT配置类
 * @version: v1.0.0
 * @author: ttao
 * @date: 2022/10/18 14:58
 */
@Configuration
public class MqttFactory {

    @Value("${mqtt.host}")
    private String host;

    @Value("${mqtt.username}")
    private String username;

    @Value("${mqtt.password}")
    private String password;

    private final static Logger logger = LoggerFactory.getLogger(MqttFactory.class);

    private static MqttFactory factory;

    private static MqttClient CLIENT;

    public MqttFactory() {
        factory = this;
    }

    public static void sendMessage(String clientId, String topic, Object message) {
        sendMessage(clientId, topic, JSONObject.toJSONString(message));
    }

    public static void sendMessage(String clientId, String topic, String message) {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(1);
        mqttMessage.setRetained(true);
        mqttMessage.setPayload(message.getBytes());
        try {
            getClient(clientId).publish(topic, mqttMessage);
            logger.info("主题【" + topic + "】发送消息成功：" + message);
            if(logger.isDebugEnabled()) {
                logger.debug("主题【" + topic + "】发送消息成功：" + message);
            }
        } catch (Exception e) {
            logger.error("主题【" + topic + "】发送消息失败：" + e.getMessage() + "\n消息内容：" + message, e);
        }

    }

    public static void subscribe(String clientId, String topic) throws MqttException {
        getClient(clientId).subscribe(topic, 1);
        logger.info("主题【" + topic + "】订阅成功");
    }

    private static MqttClient getClient(String clientId) {
        if(CLIENT != null) {
            return CLIENT;
        }
        synchronized(MqttFactory.class) {
            if(CLIENT != null) {
                return CLIENT;
            }
            try {
                CLIENT = createClient(clientId);
            } catch (MqttException e) {
                throw new RuntimeException(e.getMessage(), e);
            }
            return CLIENT;
        }
    }

    private static MqttClient createClient(String clientId) throws MqttException {
        MqttClient CLIENT = new MqttClient(factory.host, clientId, new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();

        //设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(true);

        //设置用户名密码
        options.setUserName(factory.username);
        options.setPassword(factory.password.toCharArray());

        // 设置超时时间 单位为秒
        options.setConnectionTimeout(10);

        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        options.setKeepAliveInterval(60);

        //设置断开后重新连接
        options.setAutomaticReconnect(true);

        CLIENT.connect(options);

        return CLIENT;
    }
}