Spring 集成 MQTT 推送消息

時間 2019-11-16

原文原文鏈接

(mqtt java客戶端的使用參看這篇文章)java

Spring Integration基於Eclipse Paho MQTT客戶端庫提供了支持MQTT協議的入站和出站通道適配器。可是使用起來不是很靈活(有多是沒有深刻理解😂)。在參考了這篇文章後有了以下實現：git

1. maven依賴

<dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-core</artifactId>
        <version>4.3.9.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-mqtt</artifactId>
        <version>4.3.9.RELEASE</version>
    </dependency>
    <dependency>
	    <groupId>org.eclipse.paho</groupId>
	    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
	    <version>1.2.0</version>
    </dependency>
複製代碼

2. 配置文件

applicationContext.xml

類型	描述
DefaultMqttPahoClientFactory	客戶端工廠類,根據配置的選項(用戶名、密碼、服務器集羣地址等)建立一個默認的客戶端。
MqttPahoMessageHandler	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>
複製代碼

config.properties

#用戶名
mqtt.username=mqttPubClient
#密碼
mqtt.password=123456
#是否清除會話
mqtt.cleanSession=false
#服務端url
mqtt.serverURI1=tcp://localhost:1883
#是否異步發送
mqtt.async=true
#超時時間
mqtt.completionTimeout=20000
#心跳
mqtt.keepAliveInterval=30
#客戶端id
mqtt.clientId=mqttPubClient
#默認的消息服務質量
mqtt.defaultQos=1

複製代碼

3. 調用

@Resource  
    private MqttPahoMessageHandler mqttHandler;  
	
	@Override
	public void send(String topic, String content) {
		// 構建消息
		Message<String> messages = MessageBuilder.withPayload(content).setHeader(MqttHeaders.TOPIC, topic).build();  
		// 發送消息
        mqttHandler.handleMessage(messages);
	}
複製代碼

示例代碼：github.com/brandonbai/…github