SpringBoot2.0集成MQTT功能之消息订阅处理

SpringBoot2.0集成MQTT消息之消息订阅处理

第一，pom配置，引入相关jar:




<dependency>



    <groupId>org.springframework.integration</groupId>




    <artifactId>spring-integration-stream</artifactId>




</dependency>



<dependency>



   <groupId>org.springframework.integration</groupId>




   <artifactId>spring-integration-mqtt</artifactId>





</dependency>

第二，配置MQTT服务器基本信息，在springBoot配置文件application.properties中配置，添加如下：




#MQTT配置信息



#MQTT-用户名



spring.mqtt.username=admin



#MQTT-密码



spring.mqtt.password=password



#MQTT-服务器连接地址，如果有多个，用逗号隔开，如：tcp://127.0.0.1:61613，tcp://192.168.2.133:61613



spring.mqtt.url=tcp://127.0.0.1:61613



#MQTT-连接服务器默认客户端ID



spring.mqtt.client.id=mqttId



#MQTT-默认的消息推送主题，实际可在调用接口时指定



spring.mqtt.default.topic=topic



#连接超时



spring.mqtt.completionTimeout=3000

第三，配置MQTT消息接收处理类：




/**



 * 〈一句话功能简述〉<br> 



 * 〈MQTT接收消息处理〉



 *



 * @author lenovo



 * @create 2018/6/4



 * @since 1.0.0



 */



@Configuration



@IntegrationComponentScan



public class MqttReceiveConfig {



 



    @Value("${spring.mqtt.username}")



    private String username;



 



    @Value("${spring.mqtt.password}")



    private String password;



 



    @Value("${spring.mqtt.url}")



    private String hostUrl;



 



    @Value("${spring.mqtt.client.id}")



    private String clientId;



 



    @Value("${spring.mqtt.default.topic}")



    private String defaultTopic;



 



    @Value("${spring.mqtt.completionTimeout}")



    private int completionTimeout ;   //连接超时



 



 



    @Bean



    public MqttConnectOptions getMqttConnectOptions(){



        MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();



        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(getMqttConnectOptions());



        return factory;



    }



 



    //接收通道



    @Bean



    public MessageChannel mqttInputChannel() {



        return new DirectChannel();



    }



 



    //配置client,监听的topic 



    @Bean



    public MessageProducer inbound() {



        MqttPahoMessageDrivenChannelAdapter adapter =



                new MqttPahoMessageDrivenChannelAdapter(clientId+"_inbound", mqttClientFactory(),



                        "hello","hello1");



        adapter.setCompletionTimeout(completionTimeout);



        adapter.setConverter(new DefaultPahoMessageConverter());



        adapter.setQos(1);



        adapter.setOutputChannel(mqttInputChannel());



        return adapter;



    }



 



    //通过通道获取数据



    @Bean



    @ServiceActivator(inputChannel = "mqttInputChannel")



    public MessageHandler handler() {



        return new MessageHandler() {



            @Override



            public void handleMessage(Message<?> message) throws MessagingException {




                String topic = message.getHeaders().get("mqtt_receivedTopic").toString();



                String type = topic.substring(topic.lastIndexOf("/")+1, topic.length());



                if("hello".equalsIgnoreCase(topic)){



                    System.out.println("hello,fuckXX,"+message.getPayload().toString());



                }else if("hello1".equalsIgnoreCase(topic)){



                    System.out.println("hello1,fuckXX,"+message.getPayload().toString());



                }



            }



        };



    }



}

第四，启动服务测试，使用postment调用上一篇的MQTT发送接口，分别往hello,hello1两个topic发送消息，测试接收情况：

SpringBoot2.0集成MQTT功能之消息订阅处理

由此看出，可以正常监听topic并接收处理消息了。

看到这里，朋友们可能有疑问，如果我要配置多个client，应该怎么处理呢？这个也简单，我们只要配置多个通道即可，简单代码如下：




//通道2



@Bean



public MessageChannel mqttInputChannelTwo() {



    return new DirectChannel();



}



//配置client2，监听的topic:hell2,hello3



@Bean



public MessageProducer inbound1() {



    MqttPahoMessageDrivenChannelAdapter adapter =



            new MqttPahoMessageDrivenChannelAdapter(clientId+"_inboundTwo", mqttClientFactory(),



                    "hello2","hello3");



    adapter.setCompletionTimeout(completionTimeout);



    adapter.setConverter(new DefaultPahoMessageConverter());



    adapter.setQos(1);



    adapter.setOutputChannel(mqttInputChannelTwo());



    return adapter;



}



 



//通过通道2获取数据



@Bean



@ServiceActivator(inputChannel = "mqttInputChannelTwo")



public MessageHandler handlerTwo() {



    return new MessageHandler() {



        @Override



        public void handleMessage(Message<?> message) throws MessagingException {




            String topic = message.getHeaders().get("mqtt_receivedTopic").toString();



            String type = topic.substring(topic.lastIndexOf("/")+1, topic.length());



            if("hello2".equalsIgnoreCase(topic)){



                System.out.println("hello2 clientTwo,"+message.getPayload().toString());



            }else if("hello3".equalsIgnoreCase(topic)){



                System.out.println("hello3 clientTwo,"+message.getPayload().toString());



            }



        }



    };



}

这样一来，我们就配置了两个client，client1监听处理hello、hello1主题消息，client2监听处理hello2、hello3主题，测试一下：

SpringBoot2.0集成MQTT功能之消息订阅处理

从输出结果可以看出，我们发送不同的消息，分别由不同的client处理。所以，小伙伴，你理解了吗？

【转载请注明出处——大道迷途】

SpringBoot2.0集成MQTT功能之消息订阅处理

相关推荐