【项目实战】使用 Eclipse Paho Java 让Java应用程序中实现MQTT通信协议

蔚落 2024-03-17 17:44 10阅读 0赞

## 一、Eclipse Paho Java 介绍 ##

Eclipse Paho Java 是一个开源的，功能强大，易于使用 的 MQTT 客户端库  
它提供了一个轻量级的客户端，用于在Java应用程序中实现MQTT通信协议的功能。  
它可帮助 Java 开发人员快速实现 MQTT 通信协议的功能。  
它提供多个 API，简化 MQTT 协议的实现和使用，并支持多种 MQTT 版本。  
使用Eclipse Paho Java可以轻松创建MQTT客户端，发送或接收消息。

如果您需要在客户端使用 MQTT 协议，可以选择使用 Paho。Paho 是 MQTT 客户端，而不是 MQTT 服务器。Paho 是 Eclipse 基金会开发的一个 MQTT 客户端库，可以用于各种编程语言和平台。它提供了一系列的功能和 API，使得开发人员可以轻松地使用 MQTT 协议进行开发。它支持多种消息传递模式（如发布/订阅、点对点等），还提供了安全连接、QoS 控制等高级功能。

## 二、Eclipse Paho Java的功能 ##

使用它，可以轻松创建 MQTT 客户端

Eclipse Paho Java提供了以下功能：

*  发布和订阅消息。
 *  处理连接、订阅和取消订阅等操作。
 *  提供多个API简化MQTT协议的实现和使用。  
    \-【高级功能】 提供自动重连和SSL安全连接。

Eclipse Paho Java客户端可以在不同的操作系统和设备上进行部署，例如桌面应用程序、移动应用程序和嵌入式设备。如果您需要在Java应用程序中实现MQTT通信协议的功能，Eclipse Paho Java是一个非常不错的选择。

## 三、Eclipse Paho Java的使用 ##

> 因为Eclipse Paho Java 是 MQTT 客户端 ，因此此处忽略MQTT 服务端的搭建指引。MQTT 服务端的搭建指引见文章

### 3.1 安装 MQTT 依赖 ###

添加依赖到 pom.xml

<dependency>
       <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter</artifactId>
     </dependency>
    
     <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-test</artifactId>
         <scope>test</scope>
     </dependency>
    
     <dependency>
         <groupId>org.eclipse.paho</groupId>
         <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
         <version>1.2.5</version>
     </dependency>

### 3.2 SampleCallback ###

public class SampleCallback implements MqttCallback {
        
        // 连接丢失
        @Override
        public void connectionLost(Throwable cause) {
        
            System.out.println("connection lost：" + cause.getMessage());
        }
    
        //  收到消息
        @Override
        public void messageArrived(String topic, MqttMessage message) {
        
            System.out.println("Received message: \n  topic：" + topic + "\n  Qos：" + message.getQos() + "\n  payload：" + new String(message.getPayload()));
        }
    
        // 消息传递成功
        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
        
            System.out.println("deliveryComplete");
        }
    }

### 3.3 MqttClientSample ###

以下代码演示的是，通过 TCP 端口连接，本地的 MQTT 服务器设置了账号密码为burt / burt。

public class MqttClientSample {
        
        public static void main(String[] args) {
        
            String topic = "/test/123";
            String content = "你好，我的名字叫顺溜123";
            int qos = 2;
            String broker = "tcp://0.0.0.0:1883";
            String clientId = MqttClient.generateClientId();
            //  持久化
            MemoryPersistence persistence = new MemoryPersistence();
            // MQTT 连接选项
            MqttConnectOptions connOpts = new MqttConnectOptions();
            // 设置认证信息
            connOpts.setUserName("burt");
            connOpts.setPassword("burt".toCharArray());
    
            // ssl/tls 配置
            try {
        
                // broker = "ssl://broker.emqx.io:8883";
    
                // 单向ssl/tls
                // String caCrtFile = MqttSample.class.getResource("").getPath() + "./broker.emqx.io-ca.crt";
                // connOpts.setSocketFactory(SSLUtils.getSingleSocketFactory(caCrtFile));
    
                // 双向ssl/tls
                // String caCrtFile = MqttSample.class.getResource("").getPath() + "./server-ca.crt";
                // String crtFile = MqttSample.class.getResource("").getPath() + "./client.crt";
                // String keyFile = MqttSample.class.getResource("").getPath() + "./client.key";
                // connOpts.setSocketFactory(SSLUtils.getSocketFactory(caCrtFile, crtFile, keyFile, ""));
            } catch (Exception e) {
        
                throw new RuntimeException(e);
            }
    
            try {
        
                MqttClient client = new MqttClient(broker, clientId, persistence);
                // 设置回调
                client.setCallback(new SampleCallback());
                // 建立连接
                System.out.println("Connecting to broker: " + broker);
                client.connect(connOpts);
                System.out.println("Connected to broker: " + broker);
                // 订阅 topic
                client.subscribe(topic, qos);
                System.out.println("Subscribed to topic: " + topic);
                // 发布消息
                MqttMessage message = new MqttMessage(content.getBytes());
                message.setQos(qos);
                client.publish(topic, message);
                System.out.println("Message published");
                //client.disconnect();
                //System.out.println("Disconnected");
                //client.close();
                //System.exit(0);
            } catch (MqttException me) {
        
                System.out.println("reason " + me.getReasonCode());
                System.out.println("msg " + me.getMessage());
                System.out.println("loc " + me.getLocalizedMessage());
                System.out.println("cause " + me.getCause());
                System.out.println("excep " + me);
                me.printStackTrace();
            }
        }
    }

运行之后，可以在控制台中看到输出，控制台输出如下：  
我们可以看到已经成功连接，成功订阅，发布，并且收到发布的消息。  
![在这里插入图片描述][320d700410e041308bcef18b3dc640ce.png]

[320d700410e041308bcef18b3dc640ce.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/2023ae2c476742689092fcc44c19a580.png