3. MQTT over TCP桥接

任务说明

本案例实现一个MQTT网关，通过TCP协议连接远程MQTT Broker，将本地MQTT客户端发布的消息转发至远程Broker，同时将远程Broker下发的消息回传至本地客户端。网关作为本地客户端与远程Broker之间的透明桥接层。

环境依赖

需要安装 paho-mqtt 客户端库：

pip install paho-mqtt

完整实现

import paho.mqtt.client as mqtt
import threading
import time

# ========== 配置 ==========
LOCAL_BROKER_HOST = "127.0.0.1"
LOCAL_BROKER_PORT = 1883
REMOTE_BROKER_HOST = "192.168.1.100"  # 远程Broker地址
REMOTE_BROKER_PORT = 1883
BRIDGE_TOPIC_PREFIX = "bridge/"  # 桥接主题前缀，避免循环转发

# ========== 远程Broker客户端 ==========
remote_client = mqtt.Client(client_id="bridge_remote")

def on_remote_connect(client, userdata, flags, rc):
    print(f"[远程] 已连接远程Broker, 结果码: {rc}")
    # 订阅远程下发的主题
    client.subscribe(f"{BRIDGE_TOPIC_PREFIX}#")

def on_remote_message(client, userdata, msg):
    """远程Broker消息 -> 转发到本地Broker"""
    topic = msg.topic[len(BRIDGE_TOPIC_PREFIX):]  # 去掉前缀
    local_client.publish(topic, msg.payload, msg.qos, retain=msg.retain)
    print(f"[远程->本地] topic={topic}, payload={msg.payload.decode()}")

remote_client.on_connect = on_remote_connect
remote_client.on_message = on_remote_message

# ========== 本地Broker客户端 ==========
local_client = mqtt.Client(client_id="bridge_local")

def on_local_connect(client, userdata, flags, rc):
    print(f"[本地] 已连接本地Broker, 结果码: {rc}")
    # 订阅所有本地消息（排除桥接回环主题）
    client.subscribe("#")

def on_local_message(client, userdata, msg):
    """本地Broker消息 -> 转发到远程Broker"""
    # 防止循环：不转发来自远程的消息（通过前缀识别）
    if msg.topic.startswith(BRIDGE_TOPIC_PREFIX):
        return
    remote_topic = f"{BRIDGE_TOPIC_PREFIX}{msg.topic}"
    remote_client.publish(remote_topic, msg.payload, msg.qos, retain=msg.retain)
    print(f"[本地->远程] topic={msg.topic}, payload={msg.payload.decode()}")

local_client.on_connect = on_local_connect
local_client.on_message = on_local_message

# ========== 启动 ==========
def start_bridge():
    # 连接远程Broker
    remote_client.connect(REMOTE_BROKER_HOST, REMOTE_BROKER_PORT, keepalive=60)
    remote_client.loop_start()

    # 连接本地Broker
    local_client.connect(LOCAL_BROKER_HOST, LOCAL_BROKER_PORT, keepalive=60)
    local_client.loop_start()

    print("MQTT over TCP 桥接已启动")
    print(f"本地: {LOCAL_BROKER_HOST}:{LOCAL_BROKER_PORT}")
    print(f"远程: {REMOTE_BROKER_HOST}:{REMOTE_BROKER_PORT}")
    print(f"桥接主题前缀: {BRIDGE_TOPIC_PREFIX}")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n正在停止桥接...")
        local_client.loop_stop()
        remote_client.loop_stop()
        local_client.disconnect()
        remote_client.disconnect()

if __name__ == "__main__":
    start_bridge()

运行与验证

前置条件：本地和远程各有一个MQTT Broker在运行（如Mosquitto）。修改代码中的 REMOTE_BROKER_HOST 为实际远程Broker地址。

启动桥接：

python mqtt_tcp_bridge.py

验证步骤：

1. 在远程Broker所在机器上订阅主题 bridge/test：
   mosquitto_sub -h 远程IP -t "bridge/test" -v
2. 在本地Broker上发布消息：
   mosquitto_pub -h 127.0.0.1 -t "test" -m "hello remote"
3. 观察远程订阅端应收到 bridge/test hello remote，同时桥接程序打印 [本地->远程] topic=test
4. 反向验证：在远程发布 bridge/back 主题消息，本地订阅 back 主题应能收到。

通过条件：桥接程序控制台无报错，双向消息均能按上述路径正确转发，且本地发布普通主题不会在本地产生回环重复。

关键说明

• 使用 bridge/ 前缀区分消息来源方向，避免桥接两端互相订阅导致无限循环转发。
• 本地客户端订阅 # 捕获所有本地消息；远程客户端订阅 bridge/# 接收需转发到本地的消息。
• QoS和retain标志在转发时原样传递，保证消息语义不丢失。
• 本实现为最小可用桥接，未处理断线重连与持久会话；生产环境可在此基础上增加重连逻辑。