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

    最近做了一个小项目，简单描述下项目结构，主要是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

项目结构如下图所示

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

1、引入jar包

        <!-- mqtt开始 -->
		<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>
		<!-- mqtt结束 -->

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
#u662Fu5426u6E05u9664u4F1Au8BDD
mqtt.cleanSession=false
#u662Fu5426u5F02u6B65u53D1u9001
mqtt.async=true
#u5FC3u8DF3
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进行接口测试：

查看消息队列

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

查看控制台

控制台成功打印出消息

项目已经打包上传，打包地址：https://download.csdn.net/download/qq_34178998/11251538

最后

以上就是无语画板为你收集整理的SpringMVC+Spring+Mybatis框架整合MQTT通信协议+ActiveMQ作为中间件进行消息的发布与订阅的全部内容，希望文章能够帮你解决SpringMVC+Spring+Mybatis框架整合MQTT通信协议+ActiveMQ作为中间件进行消息的发布与订阅所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。