Springboot整合mybatisPlus+mysql+druid+swaggerUI+ mqtt 整合mqtt整合druid整合mybatis-plus完整pom完整yml整合swaggerUi整合log4j MQTT 物联网系统基本架构本物联网系列
mqtt)
org.springframework.boot spring-boot-starter-integration org.springframework.integration spring-integration-stream org.springframework.integration spring-integration-mqtt
yml
iot:
mqtt:
clientId: ${random.value}
defaultTopic: topic
shbykjTopic: shbykj_topic
url: tcp://127.0.0.1:1883
username: admin
password: admin
completionTimeout: 3000
package com.shbykj.handle.mqtt;
import lombok.Getter;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.stereotype.Component;
/**
* @Author: wxm
* @Description: mqtt基础配置类
*/
@Getter
@Setter
@Component
@IntegrationComponentScan
@ConfigurationProperties(prefix = "iot.mqtt")
public class BykjMqttConfig {
/*
*
* 服务地址
*/
private String url;
/**
* 客户端id
*/
private String clientId;
/*
*
* 默认主题
*/
private String defaultTopic;
/*
*
* 用户名和密码*/
private String username;
private String password;
/**
* 超时时间
*/
private int completionTimeout;
/**
* shbykj自定义主题
*/
private String shbykjTopic;
}
package com.shbykj.handle.mqtt.producer;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
/**
* @description rabbitmq mqtt协议网关接口
* @date 2020/6/8 18:26
*/
@MessagingGateway(defaultRequestChannel = "iotMqttInputChannel")
public interface IotMqttGateway {
void sendMessage2Mqtt(String data);
void sendMessage2Mqtt(String data, @Header(MqttHeaders.TOPIC) String topic);
void sendMessage2Mqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
}
package com.shbykj.handle.mqtt;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
@Configuration
public class IotMqttProducerConfig {
public final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private BykjMqttConfig mqttConfig;
/*
*
* MQTT连接器选项
* *
*/
@Bean(value = "getMqttConnectOptions")
public MqttConnectOptions getMqttConnectOptions1() {
MqttConnectOptions mqttCOnnectOptions= new MqttConnectOptions();
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
mqttConnectOptions.setCleanSession(true);
// 设置超时时间 单位为秒
mqttConnectOptions.setConnectionTimeout(mqttConfig.getCompletionTimeout());
mqttConnectOptions.setAutomaticReconnect(true);
mqttConnectOptions.setUserName(mqttConfig.getUsername());
mqttConnectOptions.setPassword(mqttConfig.getPassword().toCharArray());
mqttConnectOptions.setServerURIs(new String[]{mqttConfig.getUrl()});
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
mqttConnectOptions.setKeepAliveInterval(10);
// 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
//mqttConnectOptions.setWill("willTopic", WILL_DATA, 2, false);
return mqttConnectOptions;
}
/**
* mqtt工厂
*
* @return
*/
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
// factory.setServerURIs(mqttConfig.getServers());
factory.setConnectionOptions(getMqttConnectOptions1());
return factory;
}
@Bean
public MessageChannel iotMqttInputChannel() {
return new DirectChannel();
}
// @Bean
// @ServiceActivator(inputChannel = "iotMqttInputChannel")
// public MessageHandler mqttOutbound() {
// MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfig.getClientId(), mqttClientFactory());
// messageHandler.setAsync(false);
// messageHandler.setDefaultQos(2);
// messageHandler.setDefaultTopic(mqttConfig.getDefaultTopic());
// return messageHandler;
// }
@Bean
@ServiceActivator(inputChannel = "iotMqttInputChannel")
public MessageHandler handlerTest() {
return message -> {
try {
String string = message.getPayload().toString();
System.out.println(string);
} catch (MessagingException ex) {
ex.printStackTrace();
logger.info(ex.getMessage());
}
};
}
}
package com.shbykj.handle.mqtt;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
/**
* @Author: xiaofu
* @Description: 消息订阅配置
* @date 2020/6/8 18:24
*/
@Configuration
public class IotMqttSubscriberConfig {
public final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private MqttReceiveHandle mqttReceiveHandle;
@Autowired
private BykjMqttConfig mqttConfig;
/*
*
* MQTT连接器选项
* *
*/
@Bean(value = "getMqttConnectOptions")
public MqttConnectOptions getMqttConnectOptions1() {
MqttConnectOptions mqttCOnnectOptions= new MqttConnectOptions();
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
mqttConnectOptions.setCleanSession(true);
// 设置超时时间 单位为秒
mqttConnectOptions.setConnectionTimeout(10);
mqttConnectOptions.setAutomaticReconnect(true);
// mqttConnectOptions.setUserName(mqttConfig.getUsername());
// mqttConnectOptions.setPassword(mqttConfig.getPassword().toCharArray());
mqttConnectOptions.setServerURIs(new String[]{mqttConfig.getUrl()});
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
mqttConnectOptions.setKeepAliveInterval(10);
// 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
//mqttConnectOptions.setWill("willTopic", WILL_DATA, 2, false);
return mqttConnectOptions;
}
/*
*
*MQTT信息通道（生产者）
**
*/
@Bean
public MessageChannel iotMqttOutboundChannel() {
return new DirectChannel();
}
/*
*
*MQTT消息处理器（生产者）
**
*/
@Bean
@ServiceActivator(inputChannel = "iotMqttOutboundChannel")
public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfig.getClientId(), mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(mqttConfig.getDefaultTopic());
return messageHandler;
}
/*
*
*MQTT工厂
**
*/
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
// factory.setServerURIs(mqttConfig.getServers());
factory.setConnectionOptions(getMqttConnectOptions1());
return factory;
}
/*
*
*MQTT信息通道（消费者）
**
*/
@Bean
public MessageChannel iotMqttInputChannel() {
return new DirectChannel();
}
/**
* 配置client,监听的topic
* MQTT消息订阅绑定（消费者）
***/
@Bean
public MessageProducer inbound() {
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId(), mqttClientFactory(), mqttConfig.getDefaultTopic(), mqttConfig.getShbykjTopic());
adapter.setCompletionTimeout(mqttConfig.getCompletionTimeout());
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(2);
adapter.setOutputChannel(iotMqttInputChannel());
return adapter;
}
/**
* @author wxm
* @description 消息订阅
* @date 2020/6/8 18:20
*/
@Bean
@ServiceActivator(inputChannel = "iotMqttInputChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<&＃63;> message) throws MessagingException {
//处理接收消息
try {
mqttReceiveHandle.handle(message);
} catch (Exception e) {
logger.warn("消息处理异常"+e.getMessage());
e.printStackTrace();
}
}
};
}
}
package com.shbykj.handle.mqtt;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.shbykj.handle.common.DataCheck;
import com.shbykj.handle.common.RedisKey;
import com.shbykj.handle.common.RedisUtils;
import com.shbykj.handle.common.constants.Constants;
import com.shbykj.handle.common.model.ShbyCSDeviceEntity;
import com.shbykj.handle.common.model.sys.SysInstrument;
import com.shbykj.handle.resolve.mapper.SysInstrumentMapper;
import com.shbykj.handle.resolve.util.DateUtils;
import com.shbykj.handle.resolve.util.ShbyCSDeviceUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.BidiMap;
import org.apache.commons.collections.bidimap.DualHashBidiMap;
import org.apache.commons.lang3.StringUtils;
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.beans.factory.annotation.Value;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
/*
*
* mqtt客户端消息处理类
* **/
@Component
@Slf4j
@Transactional
public class MqttReceiveHandle implements MqttCallback {
private static final Logger logger = LoggerFactory.getLogger(MqttReceiveHandle.class);
@Value("${shbykj.checkCrc}")
private boolean checkcrc;
@Autowired
private SysInstrumentMapper sysInstrumentMapper;
@Autowired
private RedisUtils redisUtils;
public static BidiMap bidiMap = new DualHashBidiMap();
//记录bykj协议内容
public static Map> devMap = new HashMap();
//记录上限数量
// public static Map ctxMap = new HashMap();
public void handle(Message<&＃63;> message) {
try {
logger.info("{},客户端号：{},主题：{}，QOS:{}，消息接收到的数据：{}", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()), message.getHeaders().get(MqttHeaders.ID), message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC), message.getHeaders().get(MqttHeaders.RECEIVED_QOS), message.getPayload());
//处理mqtt数据
this.handle(message.getPayload().toString());
} catch (Exception e) {
e.printStackTrace();
log.error("处理错误" + e.getMessage());
}
}
private void handle(String str) throws Exception {
boolean flag = this.dataCheck(str);
if (flag) {
ShbyCSDeviceEntity shbyCSDeviceEntity = ShbyCSDeviceUtils.convertToSysInstrumentEntity(str);
String deviceNumber = shbyCSDeviceEntity.getPN();
String smpId = shbyCSDeviceEntity.getSMP_ID();
String smpName = shbyCSDeviceEntity.getSMP_NAME();
String smpWt = shbyCSDeviceEntity.getSMP_WT();
if (StringUtils.isEmpty(smpId) || StringUtils.isEmpty(smpName) || StringUtils.isEmpty(smpWt)) {
log.error("过滤无实际作用报文信息", str);
logger.error("过滤无实际作用报文信息", str);
return;
}
//判断设备id是否存在数据库中，存在才进行数据部分处理
//不存在就提醒需要添加设备：
QueryWrapper wrapper = new QueryWrapper();
wrapper.eq("number", deviceNumber);
wrapper.eq("is_deleted", Constants.NO);
SysInstrument sysInstrument = sysInstrumentMapper.selectOne(wrapper);
if (null == sysInstrument) {
log.error("碳氧仪不存在或已删除，设备号：{}", deviceNumber);
logger.error("碳氧仪不存在或已删除，设备号：{}", deviceNumber);
return;
}
try {
//增加实时数据
String instrumentId = sysInstrument.getId().toString();
String realDataKey = RedisKey.CSdevice_DATA_KEY + instrumentId;
this.redisUtils.set(realDataKey, shbyCSDeviceEntity);
System.out.println(shbyCSDeviceEntity);
//通讯时间
String OnlineTime= "shbykj_mqtt:onlines:" + instrumentId;
this.redisUtils.set(onlineTime, shbyCSDeviceEntity.getDataTime(), (long) Constants.RedisTimeOut.REAL_TIME_OUT);
log.info("实时数据已经更新：设备主键id" + instrumentId);
logger.info("{} 实时数据已经更新：设备主键id:{}",new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()),instrumentId);
} catch (Exception var1) {
log.error("redis处理实时报文数据逻辑异常 ：" + var1.getMessage());
logger.error("redis处理实时报文数据逻辑异常 ：" + var1.getMessage());
}
}
}
private boolean dataCheck(String message) {
boolean flag = DataCheck.receiverCheck(message);
if (!flag) {
return false;
} else {
int i = message.indexOf("QN=");
if (i <0) {
log.warn("数据包中没有QN号码: " + message);
logger.warn("数据包中没有QN号码: " + message);
return false;
} else {
i = message.indexOf("PN=");
if (i <0) {
log.warn("数据包中没有PN号码: " + message);
logger.warn("数据包中没有PN号码: " + message);
return false;
} else {
if (this.checkcrc) {
flag = DataCheck.checkCrc(message);
if (!flag) {
log.warn("crc校验失败: " + message);
logger.warn("数据包中没有PN号码: " + message);
return false;
}
}
return true;
}
}
}
}
/**
* 连接丢失
*
* @param throwable
*/
@Override
public void connectionLost(Throwable throwable) {
logger.warn("连接丢失-客户端:｛｝,原因：{}", throwable.getMessage());
}
/**
* 消息已到达
*
* @param s
* @param mqttMessage
* @throws Exception
*/
@Override
public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
}
/**
* 完成消息回调
*
* @param iMqttDeliveryToken
*/
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
}
}
pom
com.alibaba druid-spring-boot-starter 1.1.10
druid-bean.xml
京公网安备 11010802041100号