MQTT（二）Java整合MQTT

r囧r小猫 2023-10-12 21:04 79阅读 0赞

## Java整合MQTT ##

上一节知道MQTT是一个通信协议，需要一个代理服务Broker；通信设备作为客户端Client，后台系统服务器也作为客户端Client。

经过了解选用EMQX作为代理服务Broker（支持WEB界面查看）

后台服务使用Spring Integration链接EMQX

## 1.EMQX ##

简介，EMQX 是一个开源的分布式物联网 MQTT 消息服务器，它实现了 MQTT 协议的各种功能，并提供了可靠的消息传递、灵活的消息路由、可扩展的集群和高可用性等特性。EMQ X 可以作为物联网应用的消息中间件，用于连接和管理大规模的物联网设备，实现设备之间的实时通信和数据传输。

语言上，EMQ X 是使用 Erlang/OTP 编程语言实现的。Erlang 是一种通用的函数式编程语言，它具有高并发、分布式和容错性的特点，非常适合构建可扩展和可靠的分布式系统。Erlang/OTP 是一个开发框架和平台，提供了一系列工具、库和标准库，用于开发并运行 Erlang 应用程序。

EMQ X 选择使用 Erlang/OTP 的主要原因是 Erlang 在处理并发和分布式通信方面具有优秀的性能和可靠性。Erlang 提供了轻量级进程（而非操作系统线程）的并发模型，每个进程具有独立的状态和消息传递机制，可以高效地处理大量的并发连接和消息处理。此外，Erlang 的可扩展性和容错性使得 EMQ X 能够在分布式环境中实现高可用性和高性能。

Erlang/OTP 还提供了许多用于网络通信、并发控制和错误处理的库和工具，这些功能对于构建物联网 MQTT 服务器非常有用。通过使用 Erlang/OTP，EMQ X 能够提供可靠的消息传递、灵活的消息路由和高可用性等关键功能，以满足物联网应用的需求

主要特点：

1.  MQTT 协议支持：EMQ X 是一个完全兼容 MQTT 协议的消息服务器，支持 MQTT 3.1.1 和 MQTT 5.0 版本。
2.  分布式架构：EMQ X 可以在多台服务器上进行水平扩展，形成分布式集群，以处理大规模的设备连接和消息传输。
3.  高可用性：EMQ X 支持主从复制和自动故障切换，以实现高可用性和容错性，保证设备和应用程序的持续可靠性。
4.  灵活的消息路由：EMQ X 提供了丰富的消息路由功能，可以根据主题、内容、设备属性等灵活地进行消息过滤和转发。
5.  安全性和权限控制：EMQ X 支持 SSL/TLS 加密和身份验证，可以对连接和消息进行安全保护，同时提供细粒度的权限控制，确保只有授权的设备和用户可以访问特定的主题和功能。
6.  插件系统：EMQ X 提供了丰富的插件系统，可以通过插件扩展和定制功能，满足不同场景和需求的应用。
7.  实时监控和统计：EMQ X 提供了实时监控和统计功能，可以监控设备连接数、消息发布和订阅情况等指标，帮助用户了解系统运行状态和性能。
8.  多协议支持：除了 MQTT，EMQ X 还支持 AMQP、CoAP、WebSocket 等多种协议，以满足不同设备和应用的通信需求。

由于是测试关系，我选择下载WINDOWS版本

WEB登录地址：  
**http://localhost:18083/\#/login**

默认账户密码为

admin
    public

### 2.Spring Integration ###

Spring Integration 是一个基于 Spring 框架的集成框架，用于构建企业级应用程序中的消息驱动和事件驱动的系统。它提供了一组丰富的组件和工具，用于实现应用程序之间的异步消息传递、事件驱动的处理和系统集成。

Spring Integration 提供了一种声明式的方式来定义和管理消息流，它基于传统的企业集成模式（Enterprise Integration Patterns，EIP），使开发者能够通过简单的配置来实现消息的路由、转换、过滤、聚合、分割等操作。开发者可以使用 Spring Integration 提供的各种消息通道、消息适配器、消息处理器和消息端点来搭建一个灵活、可扩展和可靠的消息驱动系统。

Spring Integration 支持多种消息协议和传输方式，包括 JMS、AMQP、MQTT、HTTP、TCP、UDP 等，可以与各种消息代理、消息队列和消息中间件进行集成。它还提供了与 Spring 框架的其他模块（如 Spring Boot、Spring MVC、Spring Batch 等）的无缝集成，使得构建复杂的企业应用程序变得更加简单和高效。

Spring Integration 的主要目标是帮助开发者构建可扩展和可维护的集成解决方案，通过解耦和模块化的方式来处理异步消息和事件。它广泛应用于各种领域，包括企业集成、消息驱动的微服务架构、大数据处理等。

POM引入依赖

<dependency>
                <groupId>org.springframework.integration</groupId>
                <artifactId>spring-integration-stream</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.integration</groupId>
                <artifactId>spring-integration-mqtt</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-integration</artifactId>
            </dependency>

配置类MqttConfiguration

/**
     *  mqtt配置类
     */
    @Component
    @Configuration
    @Data
    @ConfigurationProperties("mqtt")
    public class MqttConfiguration {
        
        @Autowired
        private MqttCustomerClient mqttCustomerClient;
     
        private String host;
        private String clientId;
        private String username;
        private String password;
        private String topic;
    
        private int timeout;
        private int keepAlive;
     
        @Bean
        public MqttCustomerClient getMqttCustomerClient() {
        
            mqttCustomerClient.connect(host, clientId, username, password, timeout,keepAlive);
            // 以/#结尾表示订阅所有以test开头的主题
            mqttCustomerClient.subscribe("test/#");
            return mqttCustomerClient;
        }
    }

封装客户端Bean对象MqttCustomerClient

package com.wn.mqtt.util;
     
    import lombok.extern.slf4j.Slf4j;
    import org.eclipse.paho.client.mqttv3.*;
    import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
     
    /**
     *  mqtt客户端
     */
    @Slf4j
    @Component
    public class MqttCustomerClient {
        
        @Autowired
        private PushCallback pushCallback;
     
     
        private static MqttClient client;
     
        public  static MqttClient getClient(){
        
            return  client;
        }
     
        public static void setClient(MqttClient client){
        
            MqttCustomerClient.client=client;
        }
     
        /**
         * 客户端连接
         *
         * @param host      ip+端口
         * @param clientID  客户端Id
         * @param username  用户名
         * @param password  密码
         * @param timeout   超时时间
         * @param keeplive 保留数
         */
        public void connect(String host,String clientID,String username,String password,int timeout,int keeplive){
        
            MqttClient client;
     
            try {
        
                client=new MqttClient(host,clientID,new MemoryPersistence());
                MqttConnectOptions options=new MqttConnectOptions();
                options.setCleanSession(true);
                options.setUserName(username);
                options.setPassword(password.toCharArray());
                options.setConnectionTimeout(timeout);
                options.setKeepAliveInterval(keeplive);
                MqttCustomerClient.setClient(client);
                try {
        
                    client.setCallback(pushCallback);
                    client.connect(options);
                }catch (Exception e){
        
                    e.printStackTrace();
                }
            }catch (Exception e){
        
                e.printStackTrace();
            }
        }
     
        /**
         * 发布，默认qos为0，非持久化
         * @param topic
         * @param pushMessage
         */
        public void pushlish(String topic,String pushMessage){
        
            pushlish(0,false,topic,pushMessage);
        }
     
        /**
         * 发布
         *
         * @param qos         连接方式
         * @param retained    是否保留
         * @param topic       主题
         * @param pushMessage 消息体
         */
        public void pushlish(int qos,boolean retained,String topic,String pushMessage){
        
            MqttMessage message=new MqttMessage();
            message.setQos(qos);
            message.setRetained(retained);
            message.setPayload(pushMessage.getBytes());
            MqttTopic mqttTopic= MqttCustomerClient.getClient().getTopic(topic);
            if(null== mqttTopic){
        
                log.error("topic not exist");
            }
            MqttDeliveryToken token;
            try {
        
                token=mqttTopic.publish(message);
                token.waitForCompletion();
            }catch (MqttPersistenceException e){
        
                e.printStackTrace();
            }catch (MqttException e){
        
                e.printStackTrace();
            }
        }
     
        /**
         * 订阅某个主题，qos默认为0
         * @param topic
         */
        public void subscribe(String topic){
        
            log.error("开始订阅主题" + topic);
            subscribe(topic,0);
        }
     
        public void subscribe(String topic,int qos){
        
            try {
        
                MqttCustomerClient.getClient().subscribe(topic,qos);
            }catch (MqttException e){
        
                e.printStackTrace();
            }
        }
    }

消息监听类PushCallback

/**
     *  消费监听
     */
    @Component
    public class PushCallback implements MqttCallback {
        
     
        private static MqttClient client;
     
        @Override
        public void connectionLost(Throwable throwable) {
        
            if (client == null || !client.isConnected()) {
        
                System.out.println("连接断开，正在重连....");
            }
        }
     
        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
        
            System.out.println("接收消息主题 : " + topic);
            System.out.println("接收消息Qos : " + message.getQos());
            System.out.println("接收消息内容 : " + new String(message.getPayload()));
        }
     
        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
        
            System.out.println("deliveryComplete---------" + token.isComplete());
        }
    }

测试类

@Autowired
    private MqttCustomerClient mqttCustomerClient;
    
    @Test
    void pushlish() {
        
        for (int i = 0; i < 10; i++) {
        
            mqttCustomerClient.pushlish("test/device1", "hello mqtt............" + i);
            try {
        
                Thread.*sleep*(3000);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    }

EMQX WEB监控界面接收消息数据

![在这里插入图片描述][da0cf1396be948e2a52e03259c7a5c46.png_pic_center]

[da0cf1396be948e2a52e03259c7a5c46.png_pic_center]: https://img-blog.csdnimg.cn/da0cf1396be948e2a52e03259c7a5c46.png#pic_center