SpringMVC+Spring+Mybatis框架整合MQTT通信协议+ActiveMQ作为中间件进行消息的发布与订阅

╰+哭是因爲堅強的太久メ 2021-09-20 08:18 647阅读 0赞

最近做了一个小项目，简单描述下项目结构，主要是java与单片机进行通信，为了实现通信可以采用中间件和http的方式，但是本人因为最近在学习ActiveMQ,所以更加偏向前者，最终确定了采用消息中间件的方式进行通信。

首先完成的步骤分为这两步：

1、单片机采用MQTT的通信协议将一个消息发送到一个消息中间件（本人采用ActiveMQ,也可以采用其他的消息中间件），这里需要注意下MQTT只能采用发布/订阅的模式进行消息传输。所以单边机将会发送一个带有消息体的主题到ActiveMQ中，

2、JAVA部分则对该主题进行订阅，监听消息。

而完成这两步，我现在有两种方案可供选择：

1、单片机部分采用MQTT协议将主题消息发布到队列中，java部分直接整合ActiveMQ，用ActiveMQ对主题消息进行订阅。

2、单片机部分采用MQTT协议将主题消息发布到队列中，java部分也采用MQTT协议进行处理，整合MQTT协议。

之后经过验证，还是采用了第二种方案，理由如下：采用方案1，单片机部分是采用MQTT协议进行发送，java部分接收消息，需要对消息进行处理，采用方案2可以直接接收消息内容。

若是采用方案1，则处理消息代码如下：

public void onMessage(Message message) {
    	String responseXml = null;
    	if (message instanceof BytesMessage) {
    		ActiveMQBytesMessage bytesMessage = (ActiveMQBytesMessage) message;
    		byte[] bys = null;
    		try {
    			bys = new byte[(int) bytesMessage.getBodyLength()];
    			bytesMessage.readBytes(bys);
    			responseXml = new String(bys);
    			/*******此处为模拟数据******************/
    			System.out.println("收到32单片机消息：" + responseXml);
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	} else {
    		TextMessage bm = (TextMessage) message;
            try {
                responseXml = bm.getText();
                System.out.println("------------内容是：" + responseXml);
            } catch (JMSException e) {
                logger.error(e.getMessage(), e);
            }
    	}
    
    }

采用方案二：java部分采用SpringMVC+Spring+Mybatis框架（PS:网上SpringBoot+MQTT的资料很多）+MQTT

项目结构如下图所示

![20190621090806628.png][]

考虑项目代码可能多，所以只贴关键性代码，首先将一个ssm框架搭建好后就可以开始配置MQTT了，

1、引入jar包

<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>
    		<dependency>
    			<groupId>org.eclipse.paho</groupId>
    			<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    			<version>1.2.0</version>
    		</dependency>

2、配置spring-mqtt.xml文件

<?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  ">
    
    	
    	<bean id="propertyConfigurer"
    		class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
    		<property name="location" value="classpath:jdbc.properties" />
    	</bean>
    
    	<bean id="clientFactory"
    		class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
    		<property name="userName" value="${mqtt.username}" />
    		<property name="password" value="${mqtt.password}" />
    		<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>
    		<constructor-arg name="clientFactory" ref="clientFactory"></constructor-arg>
    		<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="CID_20190617" 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="startCaseService"
        input-channel="startCase" ref="mqttCaseService" method="startCase" />
        
        <bean id="mqttCaseService" class="com.zlw.comm.mqtt.MqttService" />
    </beans>

3、jdbc.properties配置MQTT相关参数

#-----------------------------------mqtt配置----------------------------
    #MQTT-登陆名
    mqtt.username=admin
    #MQTT-密码
    mqtt.password=admin
    #级别
    mqtt.defaultQos=1
    #MQTT-地址----配你的MQ的地址与端口
    mqtt.serverURI1=tcp://192.168.1.168:1883
    #超时时间
    mqtt.completionTimeout=20000
    #MQTT-监听的主题
    mqtt.topic=RECEIVE_DATA
    #MQTT-客户端名称
    mqtt.clientId=webClientId
    #\u662F\u5426\u6E05\u9664\u4F1A\u8BDD
    mqtt.cleanSession=false
    #\u662F\u5426\u5F02\u6B65\u53D1\u9001
    mqtt.async=true
    #\u5FC3\u8DF3
    mqtt.keepAliveInterval=30

4、消息处理类MqttService.java

public class MqttService {
    	
    	
    	public void startCase(String message){
    		JSONObject jb = FastJsonUtils.toJsonObject(message);
    		System.out.println("收到32单片机消息：" + message);
        }
    
    }

之后启动项目就可以了，当单片机部分发送主题为RECEIVE\_DATA的消息主题时候，java这边就能监听到并将消息输出，

启动项目后，这里为了方便进行调试，在java端写了一个接口，进行接口测试，接口控制层代码

@Controller
    @RequestMapping("mqtt")
    public class MqttController extends BaseController{
    	@Resource
    	private MqttPahoMessageHandler mqttHandler;
     
        /**
         * @Title: sendHello 
         * @Description: TOD
         * @author  : cxding  
         * @param mqttToSave
         * @return
         */
        @RequestMapping(value = "/hello",method = RequestMethod.POST)
        @ResponseBody
        public Result<String> sendHello(String text){
           /* JSONObject jb = FastJsonUtils.toJsonObject(text);*/
            // 构建消息
            Message<String> messages = MessageBuilder.withPayload(text)
                    .setHeader(MqttHeaders.TOPIC, "RECEIVE_DATA")
                    .build();
            // 发送消息
            mqttHandler.handleMessage(messages);
        	/*String jsonstr = FastJsonUtils.getBeanToJson(mqttToSave.getRgFlowerPot());
        	mqttGateway.sendToMqtt(jsonstr,mqttToSave.getKdTopic());*/
            return new Result<String>("OK",true);
        }
        
    }

采用postman进行接口测试：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTc4OTk4_size_16_color_FFFFFF_t_70][]

查看消息队列

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTc4OTk4_size_16_color_FFFFFF_t_70 1][]

有一条消息并且已经被消费了

查看控制台

![20190621094435680.png][]

控制台成功打印出消息

项目已经打包上传，打包地址：[https://download.csdn.net/download/qq\_34178998/11251538][https_download.csdn.net_download_qq_34178998_11251538]

[20190621090806628.png]: /images/20210708/778d146c0ac64cf09fda7b7f9f89f715.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTc4OTk4_size_16_color_FFFFFF_t_70]: /images/20210708/b29a70c62d644c6ea5b185566f4faf09.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTc4OTk4_size_16_color_FFFFFF_t_70 1]: /images/20210708/74de9903ea01443598e2175e4788a89e.png
[20190621094435680.png]: /images/20210708/5cff4671740a4f8b95601f6be8586fb2.png
[https_download.csdn.net_download_qq_34178998_11251538]: https://download.csdn.net/download/qq_34178998/11251538