在物聯網（IoT）與邊緣計算快速發展的今天，設備間的高效通信成為核心需求。MQTT 作為一種輕量級的發布 / 訂閱模式協議，憑借其低帶寬占用、強穩定性和靈活的消息路由能力，已成為物聯網通信的事實標準。無論是智能家居的設備聯動、工業傳感器的數據采集，還是車聯網的實時信息交互，MQTT 都在其中扮演著關鍵角色。
本文將從零開始搭建：從使用 Docker 部署輕量級 MQTT 服務器（Broker），到基于 Java 語言實現完整的消息發布與訂閱功能，通過清晰的步驟和可直接運行的代碼，最短時間內搭建起自己的 MQTT 通信系統。

什么是 MQTT？

MQTT（Message Queuing Telemetry Transport）是一種輕量級的發布 / 訂閱模式消息傳輸協議，專為低帶寬、不穩定網絡環境設計，廣泛應用于物聯網（IoT）、傳感器網絡和移動設備通信等場景。
核心概念

• Broker：消息服務器，負責接收和轉發所有消息
• Publisher：消息發布者，發送消息到 Broker
• Subscriber：消息訂閱者，從 Broker 接收消息
• Topic：消息主題，用于消息分類和路由
• QoS (Quality of Service)：服務質量等級，定義消息傳遞的可靠性
  -

第一步：使用 Docker 部署 MQTT Broker

我們將使用 Eclipse Mosquitto，一個流行的開源 MQTT Broker。

1. 拉取 Mosquitto 鏡像

docker pull eclipse-mosquitto

2. 創建配置文件

首先創建一個目錄用于存放配置文件和數據：

mkdir -p ~/mosquitto/config ~/mosquitto/data ~/mosquitto/log

創建配置文件 mosquitto.conf：

nano ~/mosquitto/config/mosquitto.conf

添加以下內容：

persistence true
persistence_location /mosquitto/data/
log_dest file /mosquitto/log/mosquitto.log
listener 1883
allow_anonymous true
listener 1883：MQTT 默認端口
allow_anonymous true：允許匿名連接（生產環境建議關閉）

3. 啟動 Mosquitto 容器

docker run -d \--name mosquitto \-p 1883:1883 \-v ~/mosquitto/config:/mosquitto/config \-v ~/mosquitto/data:/mosquitto/data \-v ~/mosquitto/log:/mosquitto/log \eclipse-mosquitto

4. 驗證 Broker 是否運行

docker ps | grep mosquitto

如果看到運行中的容器，說明 Broker 部署成功。

第二步：Java 客戶端實現

我們將使用 Eclipse Paho Java 客戶端庫來實現 MQTT 客戶端。

1. 添加依賴

如果使用 Maven，在pom.xml中添加：

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

2. MQTT 工具類

首先創建一個工具類封裝 MQTT 連接的通用功能：

//運行
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;public class MQTTUtils {// MQTT Broker地址private static final String BROKER = "tcp://localhost:1883";/*** 創建MQTT客戶端并連接到Broker* @param clientId 客戶端ID，應唯一* @return 已連接的MQTT客戶端* @throws MqttException 連接異常*/public static MqttClient connect(String clientId) throws MqttException {// 設置客戶端持久化方式為內存MemoryPersistence persistence = new MemoryPersistence();// 創建客戶端MqttClient client = new MqttClient(BROKER, clientId, persistence);// 配置連接選項MqttConnectOptions connOpts = new MqttConnectOptions();connOpts.setCleanSession(true); // 清除會話connOpts.setConnectionTimeout(10); // 連接超時時間connOpts.setKeepAliveInterval(20); // 心跳間隔// 連接到BrokerSystem.out.println("Connecting to broker: " + BROKER);client.connect(connOpts);System.out.println("Connected");return client;}/*** 發布消息* @param client MQTT客戶端* @param topic 消息主題* @param content 消息內容* @param qos 服務質量等級 (0, 1, 2)* @throws MqttException 發布異常*/public static void publish(MqttClient client, String topic, String content, int qos) throws MqttException {MqttMessage message = new MqttMessage(content.getBytes());message.setQos(qos);client.publish(topic, message);System.out.println("Published message: " + content + " to topic: " + topic);}/*** 訂閱主題* @param client MQTT客戶端* @param topic 要訂閱的主題* @param qos 服務質量等級* @throws MqttException 訂閱異常*/public static void subscribe(MqttClient client, String topic, int qos) throws MqttException {System.out.println("Subscribing to topic: " + topic);client.subscribe(topic, qos);}
}

3. 訂閱者客戶端實現

//運行
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;public class MQTTSubscriber {// 訂閱的主題private static final String TOPIC = "test/topic";// 客戶端IDprivate static final String CLIENT_ID = "subscriber-client";// QoS等級private static final int QOS = 1;public static void main(String[] args) {MqttClient client = null;try {// 連接到Brokerclient = MQTTUtils.connect(CLIENT_ID);// 設置消息監聽器client.setCallback(new MqttCallback() {@Overridepublic void connectionLost(Throwable cause) {System.out.println("Connection lost: " + cause.getMessage());}@Overridepublic void messageArrived(String topic, MqttMessage message) throws Exception {System.out.println("Received message on topic: " + topic);System.out.println("Message content: " + new String(message.getPayload()));System.out.println("QoS: " + message.getQos());}@Overridepublic void deliveryComplete(IMqttDeliveryToken token) {// 對于訂閱者來說，這個方法通常不需要實現}});// 訂閱主題MQTTUtils.subscribe(client, TOPIC, QOS);// 保持客戶端運行以接收消息System.out.println("Waiting for messages...");while (true) {Thread.sleep(1000);}} catch (MqttException | InterruptedException e) {System.err.println("Error: " + e.getMessage());} finally {if (client != null && client.isConnected()) {try {client.disconnect();System.out.println("Disconnected");} catch (MqttException e) {e.printStackTrace();}}}}
}

4. 發布者客戶端實現

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;public class MQTTPublisher {// 發布的主題private static final String TOPIC = "test/topic";// 客戶端IDprivate static final String CLIENT_ID = "publisher-client";// QoS等級private static final int QOS = 1;public static void main(String[] args) {MqttClient client = null;try {// 連接到Brokerclient = MQTTUtils.connect(CLIENT_ID);// 發布幾條測試消息for (int i = 1; i <= 5; i++) {String message = "Hello, MQTT! This is message " + i;MQTTUtils.publish(client, TOPIC, message, QOS);Thread.sleep(2000); // 間隔2秒發送一條}} catch (MqttException | InterruptedException e) {System.err.println("Error: " + e.getMessage());} finally {if (client != null && client.isConnected()) {try {client.disconnect();System.out.println("Disconnected");} catch (MqttException e) {e.printStackTrace();}}}}
}

第三步：運行和測試

1. 啟動訂閱者

首先運行MQTTSubscriber類，它會連接到 Broker 并開始等待接收消息：

Connecting to broker: tcp://localhost:1883
Connected
Subscribing to topic: test/topic
Waiting for messages...

2. 啟動發布者

然后運行MQTTPublisher類，它會發送 5 條消息到指定主題：

Connecting to broker: tcp://localhost:1883
Connected
Published message: Hello, MQTT! This is message 1 to topic: test/topic
Published message: Hello, MQTT! This is message 2 to topic: test/topic

3. 查看結果

在訂閱者的控制臺，你應該能看到接收到的消息：

Received message on topic: test/topic
Message content: Hello, MQTT! This is message 1
QoS: 1
Received message on topic: test/topic
Message content: Hello, MQTT! This is message 2
QoS: 1

總結
本文介紹了 MQTT 的基本概念，展示了如何使用 Docker 快速部署 Mosquitto Broker，并通過 Java 代碼實現了 MQTT 客戶端的發布和訂閱功能。