Shulou(Shulou.com)06/03 Report--

MQTT protocol is used frequently in the industrial Internet of things because of its low latency and high efficiency. The previous article introduced how to send MQTT messages with code, and this paper implements the subscription reception of MQTT messages on the basis of the previous article.

Procedure:

Introduce relevant dependency org.springframework.boot spring-boot-starter-integration org.springframework.integration spring-integration-mqtt org.projectlombok lombok true to configure MQTT server information in application.yml server: port: 8090mqtt: host: tcp://127.0.0.1:1883 clientinid: mqttinId clientoutid: mqttoutid topic: virus qoslevel: 1 # MQTT authentication username: xxx password: xxx timeout: 10000 # 20s keepalive: 20 configure MQTT message push configuration package com.favccxx.mqtt.config Import lombok.extern.slf4j.Slf4j;import org.eclipse.paho.client.mqttv3.MqttConnectOptions;import org.springframework.beans.factory.annotation.Value;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.integration.annotation.IntegrationComponentScan;import org.springframework.integration.annotation.ServiceActivator;import org.springframework.integration.channel.DirectChannel;import org.springframework.integration.core.MessageProducer;import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;import org.springframework.integration.mqtt.core.MqttPahoClientFactory Import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;import org.springframework.messaging.Message;import org.springframework.messaging.MessageChannel;import org.springframework.messaging.MessageHandler;import org.springframework.messaging.MessagingException;@Slf4j@Configuration@IntegrationComponentScanpublic class MQTTReceiveConfig {@ Value ("${mqtt.username}") private String username; @ Value ("${mqtt.password}") private String password; @ Value ("${mqtt.host}") private String hostUrl @ Value ("${mqtt.clientinid}") private String clientId; @ Value ("${mqtt.topic}") private String defaultTopic; @ Value ("${mqtt.timeout}") private int completionTimeout; / / connection timeout @ Bean public MqttConnectOptions getReceiverMqttConnectOptions () {MqttConnectOptions mqttConnectOptions=new MqttConnectOptions (); mqttConnectOptions.setCleanSession (true); mqttConnectOptions.setConnectionTimeout (10); mqttConnectOptions.setKeepAliveInterval (90) MqttConnectOptions.setAutomaticReconnect (true); mqttConnectOptions.setUserName (username); mqttConnectOptions.setPassword (password.toCharArray ()); mqttConnectOptions.setServerURIs (new String [] {hostUrl}); mqttConnectOptions.setKeepAliveInterval (2); return mqttConnectOptions;} @ Bean public MqttPahoClientFactory mqttClientFactory () {DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory (); factory.setConnectionOptions (getReceiverMqttConnectOptions ()); return factory } / / receive channel @ Bean public MessageChannel mqttInputChannel () {return new DirectChannel ();} / configure client, topic @ Bean public MessageProducer inbound () {MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter (clientId+ "_ inbound", mqttClientFactory (), defaultTopic); adapter.setCompletionTimeout (completionTimeout); adapter.setConverter (new DefaultPahoMessageConverter ()) Adapter.setQos (1); adapter.setOutputChannel (mqttInputChannel ()); return adapter } / / get data @ Bean @ ServiceActivator (inputChannel = "mqttInputChannel") public MessageHandler handler () {return new MessageHandler () {@ Override public void handleMessage (Message message) throws MessagingException {log.info ("topic: {}, data received by message: {}", message.getHeaders (). Get ("mqtt_receivedTopic"), message.getPayload ()) }}

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.