Java原生结合MQTTX---完成心跳对话（附带源码）

发布于：2025-05-11 ⋅ 阅读:(19) ⋅ 点赞:(0)

简言：✨当Java遇上MQTT：打造会"隔空传话"的魔法程序✨

导语：想不想让两个Java程序像哈利波特里的双面镜一样实时对话？今天我们将用MQTT协议+EMQX，在Ubuntu上搭建一个魔法邮局，再亲手编写会传信的Java程序！gitCode平台附赠【案例源码】🔥

源码地址：https://blog.csdn.net/huangzhe0701/article/details/145205822

参考文档：

  1. 在 Ubuntu 上安装 EMQX：https://docs.emqx.com/zh/emqx/latest/deploy/install-ubuntu.html
  2. MQTTX 下载：https://mqttx.app/zh/downloads

一、🛠️ 搭建魔法邮局（EMQX服务器）

扩展：在安装EMQX前记得先更新先软件包

apt update

1. 安装EMQX企业版
在Ubuntu终端输入以下咒语：

# 下载魔法卷轴（安装包）
wget https://www.emqx.com/zh/downloads/enterprise/5.9.0/emqx-enterprise-5.9.0-ubuntu24.04-amd64.deb




# 解开卷轴封印
sudo dpkg -i emqx-enterprise-5.9.0-ubuntu20.04-amd64.deb	




# 启动邮局服务
sudo systemctl start emqx

2. 打开魔法管理台
浏览器访问 http://localhost:18083，默认账号admin/public，你将看到：

二、📱 准备第一个信使（MQTTX客户端）

安装MQTTX桌面版

安装地址：https://mqttx.app/zh/downloads

打开后新建连接：

  • 名称：魔法邮箱_varin.cn
  • 服务器：varin:1883

三、⚡ 编写会魔法的Java程序

1. 添加咒语依赖（Maven）

<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>

2. MQTT连接核心代码

package cn.varin;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.Test;

public class MqttConnectionTest {



    /**
     *
     *
     * String serverURI,(mqtt服务端地址)
     * String clientId,(客户端id)
     * MqttClientPersistence persistence（内存持久类）
     *
     *
     */
    public static String serviceURL= "tcp://varin.cn:1883";
    public static String clientId="varya_test_01";
    public static MqttClient client;
    String user="varya";
    String password= "123456";

    static {
        try {
            // 建立一个mqqt客户端类
            client = new MqttClient(serviceURL,clientId,new MemoryPersistence());
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }


    // 创建mqtt连接
    @Test
    public void createConnectionTest() throws MqttException {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName(user);
        mqttConnectOptions.setPassword(password.toCharArray());

        client.connect(mqttConnectOptions);
//        注意：因为用test类方法执行的话，太快，可能看不出来，是否正真的建立的连接，所以添加一个死循环来保持程序的存在。
        while (true);
    }


}

2. MQTT发送消息

package cn.varin;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.Test;

public class MqttSendMessageTest {



    /**
     *
     *
     * String serverURI,(mqtt服务端地址)
     * String clientId,(客户端id)
     * MqttClientPersistence persistence（内存持久类）
     *
     *
     */
    public static String serviceURL= "tcp://varin.cn:1883";
    public static String clientId="varya_test_01";
    public static MqttClient client;
    public static String user="varya";
   public static String password= "123456";

    static {
        try {
            // 建立一个mqqt客户端类
            client = new MqttClient(serviceURL,clientId,new MemoryPersistence());
            MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
            mqttConnectOptions.setUserName(user);
            mqttConnectOptions.setPassword(password.toCharArray());
            client.connect(mqttConnectOptions);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }


    // 发送消息
    @Test
    public void SendMessageTest() throws MqttException {
        // 设置消息
        String message = "hello mqttx Client";
        MqttMessage mqttMessage = new MqttMessage(message.getBytes());
        mqttMessage.setQos(2);
        // 当一个主题的消息设置了 setRetained(true) 后，这条消息会存储在 Broker 中。如果后续有新的客户端订阅这个主题，则无论何时订阅，都会立即收到最近的一条带有 retained 属性的消息作为初始数据2。如果没有设置 retained 或者之前的消息未被保留，则新订阅者不会接收到任何历史消息。,
        mqttMessage.setRetained(true);
        // 发送消息
        client.publish("java_and_mqttx_conn", mqttMessage);

        //发送完，关闭连接
        client.disconnect();
        client.close();


    }


}

3. MQTT接收消息

package cn.varin;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.Test;

public class MqttReceiveMessageTest {



    /**
     *
     *
     * String serverURI,(mqtt服务端地址)
     * String clientId,(客户端id)
     * MqttClientPersistence persistence（内存持久类）
     *
     *
     */
    public static String serviceURL= "tcp://varin.cn:1883";
    public static String clientId="varya_test_01";
    public static MqttClient client;
    public static String user="varya";
   public static String password= "123456";

    static {
        try {
            // 建立一个mqqt客户端类
            client = new MqttClient(serviceURL,clientId,new MemoryPersistence());
            MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
            mqttConnectOptions.setUserName(user);
            mqttConnectOptions.setPassword(password.toCharArray());
            client.connect(mqttConnectOptions);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }


    // 接受消息
    @Test
    public void SendMessageTest() throws MqttException {
       client.subscribe("java_and_mqttx_conn",2);
       // 建立接收消息回调
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                // 连接丢失时调用
                System.out.println("cooection error");
            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                // 接收到消息时调用
                System.out.println("来自主题："+s);
                System.out.println("接收到的消息为："+new String(mqttMessage.getPayload()));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                // 解释消息结束时调用
                System.out.println("deliveryComplete");
            }
        });
        // 为了保持test类的连接，建立一个死循环语句
        while (true);


    }


}

四、🐉 彩蛋：会喷火的测试恐龙

参考文章：https://blog.csdn.net/huangzhe0701/article/details/145205822

在终端运行：

mosquitto_pub -h localhost -t "java_and_mqttx_conn" -m "恐龙喷火啦~🔥"

观察Java程序是否输出火焰日志！

五、💡 常见魔法失效对策

  1. 检查1883端口是否被麻瓜防火墙阻挡
  2. 确认EMQX服务像打人柳一样活跃（sudo systemctl status emqx）
  3. Java依赖是否像魔药材料一样齐全

结语：现在你的Java程序已经获得了通信魔法！快来用MQTT实现更多神奇功能吧~ 如果（程序）不显形，欢迎在评论区召唤帮忙！🎩

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布