如何用Mosquitto MQTT通过Linux Shell实现多主题订阅并分Broker转发

你的脚本无法获取消息所属主题的核心原因是：mosquitto_sub默认仅输出消息内容，不会附带主题信息。要解决这个问题，只需启用它的 verbose 模式（-v 参数），该模式下会以 [主题] [消息内容] 的格式输出每条消息，以此拆分出主题和消息内容。

修改后的Shell脚本 #

#!/bin/bash
SOURCE_BROKER="192.168.1.37"
DEST_BROKER="192.168.1.25"
USER="pi"
PASS="Pass123"

while true
do
  # 添加-v参数，让mosquitto_sub输出主题+消息的组合格式
  mosquitto_sub -h "$SOURCE_BROKER" -t "0_0" -t "0_1" -u "$USER" -P "$PASS" -v | while read -r line
  do
    # 拆分主题与消息：截取第一个空格前的内容为主题，剩余部分为消息
    topic="${line%% *}"
    message="${line#* }"
    
    echo "收到主题 ${topic} 的消息"
    # 根据主题转发到对应Broker
    if [ "$topic" = "0_0" ]; then
      mosquitto_pub -h "$SOURCE_BROKER" -t "$topic" -m "$message" -u "$USER" -P "$PASS"
    elif [ "$topic" = "0_1" ]; then
      mosquitto_pub -h "$DEST_BROKER" -t "$topic" -m "$message" -u "$USER" -P "$PASS"
    fi
  done
  sleep 10 # 断开连接后等待10秒重试
done

关键细节说明 #

• -v 参数：必须添加，这是获取主题信息的核心开关
• 消息拆分逻辑：使用${line%% *}和${line#* }拆分，能兼容消息内容包含空格的场景
• 变量包裹：所有变量用双引号包裹，避免特殊字符或空格导致的命令执行错误
• 重连机制：外层while true循环保证mosquitto_sub意外断开后自动重试

可选进阶方案（Python实现） #

如果需要处理大量主题或更复杂的转发规则，推荐使用Python的paho-mqtt库，逻辑更灵活易维护：

import paho.mqtt.client as mqtt

SOURCE_BROKER = "192.168.1.37"
DEST_BROKER = "192.168.1.25"
USER = "pi"
PASS = "Pass123"
SUBSCRIBE_TOPICS = [("0_0", 0), ("0_1", 0)]

def on_message(client, userdata, msg):
    topic = msg.topic
    message = msg.payload.decode()
    print(f"收到主题 {topic} 的消息")
    
    # 初始化发布客户端并转发
    pub_client = mqtt.Client()
    pub_client.username_pw_set(USER, PASS)
    if topic == "0_0":
        pub_client.connect(SOURCE_BROKER)
    elif topic == "0_1":
        pub_client.connect(DEST_BROKER)
    pub_client.publish(topic, message)
    pub_client.disconnect()

# 配置订阅客户端
client = mqtt.Client()
client.username_pw_set(USER, PASS)
client.on_message = on_message
client.connect(SOURCE_BROKER)
client.subscribe(SUBSCRIBE_TOPICS)

client.loop_forever()

内容的提问来源于stack exchange，提问作者Samuel Mikhael