Java原生结合MQTTX,完成心跳对话

本文涉及的产品
多模态交互后付费免费试用,全链路、全Agent
简介: 简介:本文带你用Java结合MQTT协议与EMQX服务器,在Ubuntu上实现两个程序的“隔空传话”。通过搭建消息代理、编写发送/接收代码,让Java应用实现实时通信,附完整源码与调试技巧,轻松掌握物联网通信核心技能。✨

简言:当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实现更多神奇功能吧~ 如果(程序)不显形,欢迎在评论区召唤帮忙!🎩

目录
相关文章
|
9天前
|
人工智能 运维 安全
|
7天前
|
人工智能 异构计算
敬请锁定《C位面对面》,洞察通用计算如何在AI时代持续赋能企业创新,助力业务发展!
敬请锁定《C位面对面》,洞察通用计算如何在AI时代持续赋能企业创新,助力业务发展!
|
8天前
|
机器学习/深度学习 人工智能 自然语言处理
B站开源IndexTTS2,用极致表现力颠覆听觉体验
在语音合成技术不断演进的背景下,早期版本的IndexTTS虽然在多场景应用中展现出良好的表现,但在情感表达的细腻度与时长控制的精准性方面仍存在提升空间。为了解决这些问题,并进一步推动零样本语音合成在实际场景中的落地能力,B站语音团队对模型架构与训练策略进行了深度优化,推出了全新一代语音合成模型——IndexTTS2 。
676 23
|
8天前
|
人工智能 测试技术 API
智能体(AI Agent)搭建全攻略:从概念到实践的终极指南
在人工智能浪潮中,智能体(AI Agent)正成为变革性技术。它们具备自主决策、环境感知、任务执行等能力,广泛应用于日常任务与商业流程。本文详解智能体概念、架构及七步搭建指南,助你打造专属智能体,迎接智能自动化新时代。
|
14天前
|
人工智能 JavaScript 测试技术
Qwen3-Coder入门教程|10分钟搞定安装配置
Qwen3-Coder 挑战赛简介:无论你是编程小白还是办公达人,都能通过本教程快速上手 Qwen-Code CLI,利用 AI 轻松实现代码编写、文档处理等任务。内容涵盖 API 配置、CLI 安装及多种实用案例,助你提升效率,体验智能编码的乐趣。
1113 110
|
人工智能 数据可视化 数据挖掘
Quick BI 体验&征文有奖!
瓴羊生态推出Quick BI 征文激励计划,鼓励用户分享数据分析实践经验与技术洞察,征集高质量原创文章。内容围绕AI功能体验与BI案例实践,设季奖、年奖及参与奖,优秀作者可获现金奖励、产品内测资格及官方认证形象。投稿截止至2026年3月31日。
Quick BI 体验&征文有奖!