MQTT 3 ——MQTT與Spring Mvc整合

本篇記錄一下MQTT客戶端與Spring Mvc整合

---

  網絡上已經有很多的MQTT客戶端與SpringBoot整合的技術文檔，但是與Spring Mvc框架的整合文檔似乎並不太多，可能是因為SpringMvc框架已經逐漸被淘汰了。但很幸運，我這次JAVA后端項目就是用的SpringMvc,所以在整理了網上很多資料后，也為了方便其他后來人，把我這次的整合開發過程用文字記錄一下。

POM引入依賴
  千萬注意引入依賴的版本，一定要保持3個依賴的版本一致，不同的版本可能造成各種問題。
我在開發過程中碰到因為3個版本不一致，導致Class找不到的情況，所以對版本間區別不清楚的朋友們，就不要改版本。

<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
<version>4.1.0.RELEASE</version>
</dependency>

<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>4.1.0.RELEASE</version>
</dependency>

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

 

resource目錄下添加mqtt.properties配置文件

#用戶名
mqtt.username=mqttPubClient
#密碼
mqtt.password=123456
#是否清除會話
mqtt.cleanSession=false
#服務端url
mqtt.serverURI1=tcp://127.0.0.1:1883

mqtt.async=true
#超時時間
mqtt.completionTimeout=20000
#心跳
mqtt.keepAliveInterval=30
#客戶端id
mqtt.clientId=mqttPubClient
#默認的消息服務質量
mqtt.defaultQos=1
#MQTT-監聽的主題
mqtt.topic=hello

 

寫Spring Mqtt 的配置XML
在resource目錄下添加spring-mqtt.xml,要確保這個XML能被Spring啟動時加載進去。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:int-mqtt="http://www.springframework.org/schema/integration/mqtt"
xsi:schemaLocation="
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-4.1.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://www.springframework.org/schema/integration/mqtt
http://www.springframework.org/schema/integration/mqtt/spring-integration-mqtt-4.1.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.1.xsd ">

<context:property-placeholder location="classpath:mqtt.properties" ignore-unresolvable="true"/>

<!--MQTT配置-->
<bean id="clientFactory"
class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
<property name="userName" value="${mqtt.username}"/>
<property name="password" value="${mqtt.password}"/>
<property name="cleanSession" value="${mqtt.cleanSession}"/>
<property name="keepAliveInterval" value="${mqtt.keepAliveInterval}"/>
<property name="serverURIs">
<array>
<value>${mqtt.serverURI1}</value>
</array>
</property>
</bean>

<bean id="mqttHandler" class="org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler">
<constructor-arg name="clientId" value="${mqtt.clientId}"/>
<constructor-arg name="clientFactory" ref="clientFactory"/>
<property name="async" value="${mqtt.async}"/>
<property name="defaultQos" value="${mqtt.defaultQos}"/>
<property name="completionTimeout" value="${mqtt.completionTimeout}"/>
</bean>

<!-- 消息適配器 -->
<int-mqtt:message-driven-channel-adapter
id="mqttInbound" client-id="${mqtt.clientId}" url="${mqtt.serverURI1}"
topics="${mqtt.topic}" qos="${mqtt.defaultQos}" client-factory="clientFactory" auto-startup="true"
send-timeout="${mqtt.completionTimeout}" channel="startCase" />
<int:channel id="startCase" />
<!-- 消息處理類 -->
<int:service-activator id="handlerService"
input-channel="startCase" ref="mqttCaseService" method="handler" />

<!-- 消息處理 -->
<bean id="mqttCaseService" class="com.loong.mqtt.service.impl.MqttServiceImpl" />

</beans>

 

MQTT Service 接口

public interface MqttService {

public void send(String topic,String content) throws Exception;

public void handler(String message) throws Exception;
}

 

MQTT Service 實現

import com.mqtt.service.MqttService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service("MqttService")
@Slf4j
public class MqttServiceImpl implements MqttService {

@Resource
private MqttPahoMessageHandler mqttHandler;

@Override
public void send(String topic, String content) throws Exception {
// 構建消息
Message<String> messages =
MessageBuilder.withPayload(content).setHeader(MqttHeaders.TOPIC, topic).build();
// 發送消息
mqttHandler.handleMessage(messages);
}


@Override
public void handler(String message) throws Exception {
log.info("收到消息:"+message);
}
}

 

MQTT Controller

package com.loong.mqtt.controller;

import com.loong.api.model.ApiQueryOrderModel;
import com.loong.kuaizhi.service.KuaizhiService;
import com.loong.mqtt.service.MqttService;
import com.loong.ysf.dto.Protocol;
import com.loong.ysf.model.OrderInfo;
import lombok.extern.slf4j.Slf4j;
import org.jeecgframework.core.common.service.impl.RedisService;
import org.jeecgframework.core.util.ResourceUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

/**
 */
@RestController
@RequestMapping("/mqttController")
@Slf4j
public class MqttController {
    
    @Resource(name = "MqttService")
    private MqttService mqttService;

    @RequestMapping(params = "testSend", method = RequestMethod.POST)
    @ResponseBody
    public String testSend(HttpServletRequest request, HttpServletResponse response) {
        try {
            String topic = ResourceUtil.getParameter("topic");
            String content = ResourceUtil.getParameter("content");
            this.mqttService.send(topic,content);
        } catch (Exception ex) {
            ex.printStackTrace();
            log.error("發送失敗", ex);
        }

        return "發送成功";
    }
}

 

發送消息測試

• MQTTBOX先連上MQTT服務器，並訂閱topic為hello的消息
• POSTMAN向mqttController.testSend接口發送請求，topic為hello,內容為{"hello":"200315"}
• MQTTBOX里面的訂閱者收到消息
  

接收消息測試

• MQTTBOX創建一個消息發布者，發MQTT服務器發送topic為hello,內容為{"hello":"513002"}
  

   

• 項目后台日志輸出訂閱到的內容
  

作者：admin
原文地址：www.jiansword.com