SpringBoot集成MQTT协议

简介

MQTT 可以被解释为一种低开销，低带宽占用的即时通讯协议，可以用较少的代码和带宽为远程设备连接提供实时可靠的消息服务，它适用于硬件性能低下的远程设备以及网络状况糟糕的环境下，因此 MQTT 协议在 IoT（Internet of things，物联网），小型设备应用，移动应用等方面有较广泛的应用。

优点：代码量少，开销低，带宽占用小，即时通讯协议。

MQTT原理

实现mqtt协议需要客户端和服务器端通讯完成，在通讯中，mqtt协议中有三种身份：发布者（publish），代理（broker）(服务器)，订阅者（subscribe）。其中，消息的发布者和订阅者都是客户端。消息代理是服务器，消息发布者可以同时是订阅者，传输过程如下如所示。

有别于传统的客户端/服务器通讯协议，MQTT协议并不是端到端的，消息传递通过代理，包括会话（session）也不是建立在发布者和订阅者之间，而是建立在端和代理之间。代理解除了发布者和订阅者之间的耦合。

除了发布者和订阅者之间传递普通消息，代理还可以为发布者处理保留消息和遗愿消息，并可以更改服务质量（QoS）等级。

MQTT主题

1、主题层级分隔符“/”

用于分割主题的每个层级，为主题名提供一个分层结构。如主题:

china/anhui
china/anhui/hefei

2、多层通配符“#”

用于匹配主题中任意层级的通配符。如主题:china/#

china/anhui
china/anhui/hefei
china/anhui/hefei/shushan

3、单层通配符“+”

加号是只能用于单个主题层级匹配的通配符。如主题:

china/+      只能匹配 china/anhui
china/+/+/shushan     能匹配china/anhui/hefei/shushan

4、通配符“$”

通配符“$”表示匹配一个字符，只要不是放在主题的最开头，即:

$xx
/$xx
/xx$

实战应用

对接协议部分内容

SpringBoot集成Mqtt协议

本篇文章就不重点介绍MQTT相关的内容了，主要学会怎么运用到实际开发工作中。

添加依赖

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

添加配置类

@Configuration
@ConfigurationProperties(prefix="spring.mqtt")
public class MqttProperties {private String url;private String username;private String password;private String clientId;public String getUrl() {return url;}public void setUrl(String url) {this.url = url;}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 getClientId() {return clientId;}public void setClientId(String clientId) {this.clientId = clientId;}
}

@Configuration
@ConditionalOnBean(MqttProperties.class)
public class MqttConfiguration {@Autowiredprivate MqttProperties mqttProperties;@Autowiredprivate MqttMessageHandler mqttMessageHandler;/*** 创建MqttPahoClientFactory，设置MQTT Broker连接属性，如果使用SSL验证，也在这里设置。** @return factory*/@Beanpublic MqttPahoClientFactory mqttClientFactory() {DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();MqttConnectOptions options = new MqttConnectOptions();// 设置代理端的URL地址，可以是多个options.setServerURIs(new String[]{mqttProperties.getUrl()});options.setUserName(mqttProperties.getUsername());options.setPassword(mqttProperties.getPassword().toCharArray());options.setKeepAliveInterval(120);factory.setConnectionOptions(options);return factory;}/*** 入站通道*/@Beanpublic MessageChannel mqttInputChannel() {return new DirectChannel();}/*** 入站*/@Beanpublic MessageProducer inbound() {// Paho客户端消息驱动通道适配器，主要用来订阅主题MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId() + "-consumer",mqttClientFactory(),MqttConstants.UP_DEVICE_STATE,MqttConstants.UP_DEVICE_STATUS,MqttConstants.UP_DEVICE_EVENT,MqttConstants.UP_DEVICE_SET_ACK,MqttConstants.UP_DEVICE_GET_ACK,MqttConstants.UP_DEVICE_CAPTURE_REPORT,MqttConstants.UP_DEVICE_CONTROL_QUERY_ACK,MqttConstants.UP_DEVICE_CONTROL_ACK);// Paho消息转换器DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();adapter.setConverter(defaultPahoMessageConverter);adapter.setCompletionTimeout(5000);// 设置QoSadapter.setQos(1);adapter.setOutputChannel(mqttInputChannel());return adapter;}/*** ServiceActivator注解表明：当前方法用于处理MQTT消息，inputChannel参数指定了用于消费消息的channel。** @return*/@Bean@ServiceActivator(inputChannel = "mqttInputChannel")public MessageHandler handler() {return mqttMessageHandler;}/*** 出站通道*/@Beanpublic MessageChannel mqttOutboundChannel() {return new DirectChannel();}/*** 出站*/@Bean@ServiceActivator(inputChannel = "mqttOutboundChannel")public MessageHandler outbound() {// 发送消息和消费消息Channel可以使用相同MqttPahoClientFactoryMqttPahoMessageHandler mqttPahoMessageHandler = new MqttPahoMessageHandler(mqttProperties.getClientId() + "-producer", mqttClientFactory());// 如果设置成true，即异步，发送消息时将不会阻塞。mqttPahoMessageHandler.setAsync(true);// 设置默认QoSmqttPahoMessageHandler.setDefaultQos(1);// Paho消息转换器DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();mqttPahoMessageHandler.setConverter(defaultPahoMessageConverter);return mqttPahoMessageHandler;}
}

消息处理类

@Service
public class MqttMessageHandler implements MessageHandler {private final Logger logger = LoggerFactory.getLogger(MqttMessageHandler.class);@Overridepublic void handleMessage(Message<?> message) throws MessagingException {try {String payload = message.getPayload().toString();String topic = message.getHeaders().get("mqtt_receivedTopic").toString();logger.info("接受来自mqtt的订阅信息，topic：{}", topic);//离线上报if (topic.matches(".+/offline")) {statusReport(payload);}//状态上报else if (topic.matches(".+/status")) {deviceInfoReport(payload);} //事件上报else if (topic.matches(".+/eventReport")) {eventReport(payload);} else {logger.info("主题topic:{}，负载payload:{}", topic, payload);}} catch (Exception e) {logger.error("handleMessage 接受mqtt订阅消息异常：", e);}}/*** 设备状态上报** @param payload*/private void statusReport(String payload) {OfflineReport offlineReport = JSONUtil.toBean(payload, OfflineReport.class);logger.info("收到设备状态信息上报：{}", offlineReport);//...省略}//...省略}

/*** 消息发送*/
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {/*** 定义重载方法，用于消息发送** @param payload*/void sendToMqtt(String payload);/*** 指定topic进行消息发送** @param topic* @param payload*/void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, byte[] payload);
}