MQTT - 订阅发布demo

àì夳堔傛蜴生んèń 2022-12-25 02:56 114阅读 0赞

本文中 MQTTBroker 基于 moquette 实现，本地安装 moquette 参见 [MQTT - broker - 本地安装启动moquette][MQTT - broker - _moquette]

demo基于 MQTT Cllient 实现， mqtt-client项目代码参见： [https://github.com/fusesource/mqtt-client][https_github.com_fusesource_mqtt-client]

# 一 MQTT Server #

自定义MQTTServer实现类，提供消息发布服务，即生产消息

package org.fusesource.mqtt.client;
    
    /**
     * MQTT moquette 的Server  用于发布消息至指定topic
     */
    public class MQTTServer {
    
    
        public static void main(String[] args) {
            MQTT mqtt = new MQTT();
            BlockingConnection connection = null;
            try {
                //设置服务端的ip
                mqtt.setHost("tcp://localhost:1883");
    
    //            mqtt.setHost("tcp://192.168.8.10:1883");
    //            mqtt.setUserName("test");
    //            mqtt.setPassword("test");
    
                //连接前清空会话信息
                mqtt.setCleanSession(true);
                //设置重新连接的次数
                mqtt.setReconnectAttemptsMax(6);
                //设置重连的间隔时间
                mqtt.setReconnectDelay(2000);
                //设置心跳时间 低耗网络，但是又需要及时获取数据，心跳30s
                mqtt.setKeepAlive((short)30);
                //设置缓冲的大小  最大2M
                mqtt.setSendBufferSize(2 * 1024 * 1024);
    
                //创建阻塞式连接
                connection = mqtt.blockingConnection();
                
                connection.connect();
                try {
                    int count = 0;
                    while (true) {
                        count++;
                        // 指定topic
                        String topic = "test/test1";
                        // 设置消息内容
                        String message = "hello " + count;
                        connection.publish(topic, message.getBytes(), QoS.EXACTLY_ONCE, false);
                        System.out.println("MQTTServer receive message:  Topic: " + topic + "  Content :" + message);
                        Thread.sleep(2000);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    connection.disconnect();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

# 二 MQTT Client #

自定义 MQTTCllient类，实现消息订阅服务，即消费消息

package org.fusesource.mqtt.client;
    
    /**
     * MQTT moquette 的Client 用于消费指定topic的消息
     */
    public class MQTTClient {
    
    
        public static void main(String[] args) {
            //创建MQTT对象
            MQTT mqtt = new MQTT();
            BlockingConnection connection = null;
            try {
                //设置mqtt broker的ip和端口
                mqtt.setHost("tcp://localhost:1883");
    
                // 设置用户名密码类型连接
    //            mqtt.setHost("tcp://192.168.8.10:1883");
    //            mqtt.setUserName("test");
    //            mqtt.setPassword("test");
    //            mqtt.setClientId("client_123");
    
                //连接前清空会话信息
                mqtt.setCleanSession(true);
                //设置重新连接的次数
                mqtt.setReconnectAttemptsMax(6);
                //设置重连的间隔时间 单位毫秒
                mqtt.setReconnectDelay(2000);
                //设置心跳时间  低耗网络，但是又需要及时获取数据，心跳30s
                mqtt.setKeepAlive((short)30);
                //设置缓冲的大小 最大2M
                mqtt.setSendBufferSize(2 * 1024 * 1024);
    
                //获取mqtt的连接对象BlockingConnection
                connection = mqtt.blockingConnection();
                //创建连接
                connection.connect();
                //创建相关的MQTT 的主题列表
                Topic[] topics = {new Topic("test/test1", QoS.AT_LEAST_ONCE)};
                //订阅相关的主题信息
                byte[] qoses = connection.subscribe(topics);
                //
                while (true) {
                    System.out.println("waiting...");
    
                    //接收订阅的消息内容
                    Message message = connection.receive();
    
                    System.out.println("received...");
    
                    //获取订阅的消息内容
                    byte[] payload = message.getPayload();
    
                    //
                    System.out.println("MQTTClient Message  Topic：" + message.getTopic() + " Content :" + new String(payload));
                    //确认消息回执
                    message.ack();
                    Thread.sleep(2000);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    connection.disconnect();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

# 三 本地运行 #

本地运行 MQTTServer.java，生产消息：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N4ZzAyMDU_size_16_color_FFFFFF_t_70][]

本地运行MQTTClient.java，消费消息：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N4ZzAyMDU_size_16_color_FFFFFF_t_70 1][]

[MQTT - broker - _moquette]: https://blog.csdn.net/sxg0205/article/details/110368424
[https_github.com_fusesource_mqtt-client]: https://github.com/fusesource/mqtt-client
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N4ZzAyMDU_size_16_color_FFFFFF_t_70]: /images/20221120/d6a87839ee2d473eb8c6bb4c1f4f0a2b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N4ZzAyMDU_size_16_color_FFFFFF_t_70 1]: /images/20221120/30439ac703bc43ce972c7bd86e8f485c.png