MQTT+Springboot整合

1.mqttconfig配置(配置参数是从数据库查出来的)

package com.terminal.dc3.api.center.manager.config;import com.collection.common.utils.StringUtils;
import com.collection.system.mapper.MqttConfigMapper;
import lombok.Data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;import java.util.List;
import java.util.concurrent.ConcurrentHashMap;@Component
@Data
public class MqttConfig {@Autowiredprivate MqttPushClient mqttPushClient;@Autowiredprivate MqttConfigMapper mqttConfigMapper;public static ConcurrentHashMap<String, List<MqttPushClient>> mapHashMap = new ConcurrentHashMap<>();/*** 用户名*/private String username;/*** 密码*/private String password;/*** 连接地址*/private String hostUrl;/*** 端口*/private String port;/*** 客户Id*/private String clientId;/*** 默认连接话题*/private String defaultTopic;/*** 超时时间*/private int timeout;/*** 保持连接数*/private int keepalive;/*** mqtt功能使能*/private boolean enabled;private boolean retained;/*** qos*/private int qos;public String getUsername() {return username;}public void setUsername(String username) {this.username = username;}public String getPassword() {return password;}public void setPassword(String password) {this.password = password;}public String getHostUrl() {return hostUrl;}public void setHostUrl(String hostUrl) {this.hostUrl = hostUrl;}public String getClientId() {return clientId;}public void setClientId(String clientId) {this.clientId = clientId;}public String getDefaultTopic() {return defaultTopic;}public void setDefaultTopic(String defaultTopic) {this.defaultTopic = defaultTopic;}public int getTimeout() {return timeout;}public void setTimeout(int timeout) {this.timeout = timeout;}public int getKeepalive() {return keepalive;}public void setKeepalive(int keepalive) {this.keepalive = keepalive;}public boolean isEnabled() {return enabled;}public void setEnabled(boolean enabled) {this.enabled = enabled;}public int getQos() {return qos;}public void setQos(int qos) {this.qos = qos;}//    @Beanpublic MqttPushClient getMqttPushClient() {List<com.collection.system.domain.MqttConfig> mqttConfigs = mqttConfigMapper.selectMqttConfigList(new com.collection.system.domain.MqttConfig());for (com.collection.system.domain.MqttConfig mq : mqttConfigs) {mq.setHost("tcp://" + mq.getHost() + ":" + mq.getPort());if (mq.getEnabled()) {String mqtt_topic[] = StringUtils.split(mq.getTopic(), ",");mqttPushClient.connect(mq.getHost(), mq.getClientId(), mq.getUsername(),mq.getPassword(), mq.getTimeout(), mq.getKeepalive());//连接if (mqtt_topic != null) {for (int i = 0; i < mqtt_topic.length; i++) {mqttPushClient.subscribe(mqtt_topic[i], mq.getQos().intValue());//订阅主题}}}}return mqttPushClient;}
}

2.MqttPushClient客户端配置

package com.terminal.dc3.api.center.manager.config;import com.collection.common.core.redis.RedisCache;
import com.collection.common.utils.QueueUtils;
import com.collection.common.utils.SecurityUtils;
import com.collection.common.utils.StringUtils;
import com.collection.common.utils.ip.IpUtils;
import com.collection.common.utils.uuid.IdUtils;
import com.collection.ems.domain.MqttDatasLog;
import com.collection.ems.mapper.MqttDatasLogMapper;
import com.terminal.dc3.driver.service.mqtt.PushCallback;
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.Autowired;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.Date;@Component
public class MqttPushClient {private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class);@Autowiredprivate PushCallback pushCallback;private static MqttConfig mqttConfig;@Autowiredpublic void setMqttConfig(MqttConfig mqttConfig) {this.mqttConfig = mqttConfig;}private static MqttClient client;private static MqttClient getClient() {return client;}private static void setClient(MqttClient client) {MqttPushClient.client = client;}private static MqttDatasLogMapper mqttDatasLogMapper;@Autowiredpublic void setMqttDatasLogMapper(MqttDatasLogMapper mqttDatasLogMapper) {this.mqttDatasLogMapper = mqttDatasLogMapper;}/*** 客户端连接** @param host      ip+端口* @param clientID  客户端Id* @param username  用户名* @param password  密码* @param timeout   超时时间* @param keepalive 保留数*/public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) {MqttClient client;try {if (host != null && clientID != null && username != null && password != null) {client = new MqttClient(host, clientID, new MemoryPersistence());MqttConnectOptions options = new MqttConnectOptions();options.setCleanSession(true);options.setUserName(username);options.setPassword(password.toCharArray());options.setConnectionTimeout(timeout);options.setKeepAliveInterval(keepalive);options.setAutomaticReconnect(true);MqttPushClient.setClient(client);client.setCallback(pushCallback);client.connect(options);}} catch (Exception e) {logger.error("connect error", e);}}/*** 发布消息** @param pubTopic 主题* @param message  内容* @param qos      连接方式*/public void publishMessage(String pubTopic, String message, int qos) {//重新进行连接if (client == null || !client.isConnected()) {mqttConfig.getMqttPushClient();logger.info("重新获取连接  {}" + client);}MqttMessage mqttMessage = new MqttMessage();mqttMessage.setQos(qos);mqttMessage.setPayload(message.getBytes());if (client == null) {return;}MqttTopic topic = client.getTopic(pubTopic);//记录发送日志MqttDatasLog mqttDatasLog = new MqttDatasLog();mqttDatasLog.setDataLog(IdUtils.randomUUID());mqttDatasLog.setLogType(pubTopic);mqttDatasLog.setContent(message);mqttDatasLog.setOperIp(IpUtils.getHostIp());mqttDatasLog.setOperTime(new Date());mqttDatasLog.setBusinessType(0L);if (null != topic) {try {MqttDeliveryToken publish = topic.publish(mqttMessage);if (!publish.isComplete()) {logger.info("发布消息成功");mqttDatasLog.setStatus(0l);//在队列中删除推送成功的数据QueueUtils.removeQueueObject(pubTopic, new String(mqttMessage.getPayload()));}} catch (Exception e) {logger.error("发布消息错误 error");mqttDatasLog.setStatus(1l);}}mqttDatasLogMapper.insertMqttDatasLog(mqttDatasLog);}/*** 订阅某个主题** @param topic 主题* @param qos   连接方式*/public static void subscribe(String topic, int qos) {logger.info("开始订阅主题" + topic + "连接方式" + qos);try {MqttPushClient.getClient().subscribe(topic, qos);} catch (Exception e) {logger.error("开始订阅主题错误 error");}}}

3.MQTTMessage数据类

package com.terminal.dc3.api.center.manager.config;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;import java.io.Serializable;/*** 消息实体对象**/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class MQTTMessage implements Serializable {/*** MQTT主题*/private String topic;/*** qos*/private Integer qos = 1;/*** MQTT内容*/private String content;}

4.MQTTLicenseTask(这是我的发送数据定时器，仅供参考)

package com.collection.web.controller.ems.task;import com.collection.common.queue.QueueCodeConstant;
import com.collection.common.utils.QueueUtils;
import com.collection.common.utils.StringUtils;
import com.collection.system.domain.MqttConfig;
import com.collection.system.mapper.MqttConfigMapper;
import com.terminal.dc3.api.center.manager.config.MqttPushClient;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.List;@Component
@Slf4j
public class MQTTLicenseTask {@Resourceprivate MqttConfigMapper mqttConfigMapper;@Resourceprivate MqttPushClient mqttPushClient;//    @Scheduled(fixedRate = 5000) // 每1000毫秒（1秒）执行一次@RabbitListener(queues = QueueCodeConstant.mqtt_task_config, containerFactory = "rabbitListenerContainerFactory")public void executeTask() {List<MqttConfig> mqttConfigs = mqttConfigMapper.selectMqttConfigList(new com.collection.system.domain.MqttConfig());for (com.collection.system.domain.MqttConfig mq : mqttConfigs) {String mqtt_topic[] = StringUtils.split(mq.getTopic(), ",");if (mqtt_topic != null) {for (int i = 0; i < mqtt_topic.length; i++) {Object queueObject = QueueUtils.getQueueObject(mqtt_topic[i]);if (queueObject != null) {log.info("定时推送");mqttPushClient.publishMessage(mqtt_topic[i], String.valueOf(queueObject), mq.getQos().intValue());}}}}}}