如何实现MQTT消息触发Bash脚本并传递负载执行？

方案一：Bash + mosquitto_sub 原生组合（无需额外依赖） #

直接用Bash循环结合mosquitto_sub实现持续订阅，收到消息就触发脚本，适合轻量需求。

1. 编写监听脚本 mqtt_listener.sh #

#!/bin/bash
# 配置你的MQTT参数
MQTT_BROKER="localhost"  # 替换为你的MQTT broker地址
MQTT_TOPIC="your/target/topic"  # 替换为目标主题
MQTT_USER="user"
MQTT_PASS="password"
PROCESS_SCRIPT="./process_message.sh"  # 你的处理脚本路径

# 持续订阅消息，每条消息触发处理脚本
mosquitto_sub -h "$MQTT_BROKER" -u "$MQTT_USER" -P "$MQTT_PASS" -t "$MQTT_TOPIC" | while read -r payload; do
    # 用双引号包裹参数，避免特殊字符（空格、引号等）导致的问题
    "$PROCESS_SCRIPT" "$payload"
done

2. 设置为守护进程 #

• 临时后台运行：

  chmod +x mqtt_listener.sh process_message.sh
  nohup ./mqtt_listener.sh > /dev/null 2>&1 &
  

• 开机自启（用systemd）：
  创建/etc/systemd/system/mqtt-listener.service文件：

  [Unit]
  Description=MQTT Message Listener
  After=network.target
  
  [Service]
  User=your_username
  WorkingDirectory=/path/to/your/scripts
  ExecStart=/path/to/your/scripts/mqtt_listener.sh
  Restart=always
  
  [Install]
  WantedBy=multi-user.target
  

  然后执行：

  sudo systemctl daemon-reload
  sudo systemctl enable --now mqtt-listener.service
  

方案二：Python + paho-mqtt（灵活扩展） #

如果需要后续添加额外逻辑（比如错误重试、日志记录），用Python的paho-mqtt库更方便。

1. 依赖安装 #

pip install paho-mqtt

2. 编写监听脚本 mqtt_listener.py #

import paho.mqtt.client as mqtt
import subprocess
import logging

# 配置参数
MQTT_BROKER = "localhost"
MQTT_PORT = 1883
MQTT_TOPIC = "your/target/topic"
MQTT_USER = "user"
MQTT_PASS = "password"
PROCESS_SCRIPT = "./process_message.sh"

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        logger.info("Connected to MQTT broker successfully")
        client.subscribe(MQTT_TOPIC)
    else:
        logger.error(f"Connection failed with code {rc}")

def on_message(client, userdata, msg):
    try:
        payload = msg.payload.decode("utf-8")
        logger.info(f"Received payload: {payload}")
        # 调用处理脚本，传递负载参数
        subprocess.run([PROCESS_SCRIPT, payload], check=True, capture_output=True, text=True)
        logger.info("Script executed successfully")
    except UnicodeDecodeError:
        logger.error("Failed to decode payload (not UTF-8)")
    except subprocess.CalledProcessError as e:
        logger.error(f"Script failed: {e.stderr}")

# 初始化MQTT客户端
client = mqtt.Client()
client.username_pw_set(MQTT_USER, MQTT_PASS)
client.on_connect = on_connect
client.on_message = on_message

# 连接并持续运行
try:
    client.connect(MQTT_BROKER, MQTT_PORT, 60)
    client.loop_forever()
except Exception as e:
    logger.error(f"Fatal error: {e}")

3. 后台运行 #

同样可以用nohup或者systemd配置为服务，和方案一类似。

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