MQTT 命令行操作

MQTT 命令行操作

概述

本节介绍 MQTT 命令行工具的使用方法，包括 mosquitto_pub、mosquitto_sub 和其他实用命令。学习完成后，您将能够：

• 使用命令行发布和订阅 MQTT 消息
• 使用 CLI 工具进行调试和测试
• 编写 MQTT 测试脚本
• 监控 Broker 运行状态

前置要求

在开始本节之前，请确保：

• MQTT Broker 已运行
• Mosquitto 客户端工具已安装
• 基本的命令行操作经验

mosquitto_pub

Basic Usage

# 基本语法
mosquitto_pub -h <host> -t <topic> -m <message>

# 常用参数
-h, --host     Broker 主机地址（默认 localhost）
-p, --port     端口号（默认 1883）
-t, --topic    发布 Topic
-m, --message  消息内容
-q, --qos      QoS 级别 (0/1/2)
-r, --retain   保留消息
-d, --debug    调试模式
-u, --username 用户名
-P, --password 密码
-f, --file     从文件读取消息
-n, --null-message 发送空消息

Practical Examples

# 1. 发布简单消息
mosquitto_pub -h localhost -t "test/topic" -m "Hello MQTT"

# 2. 发布 JSON 数据
mosquitto_pub -h localhost -t "sensors/temperature" \
  -m '{"value": 26.5, "unit": "celsius"}'

# 3. 指定 QoS 和保留标志
mosquitto_pub -h localhost -t "devices/status" \
  -q 1 -r -m '{"device": "esp32_001", "status": "online"}'

# 4. 从文件读取消息
echo '{"temperature": 26.5, "humidity": 62}' > data.json
mosquitto_pub -h localhost -t "sensors/environment" -f data.json

# 5. 发送空消息（常用于清除保留消息）
mosquitto_pub -h localhost -t "devices/status" -n -r

# 6. 带认证发布
mosquitto_pub -h localhost -t "secure/topic" \
  -u "admin" -P "password123" \
  -m "Authenticated message"

# 7. 调试模式查看协议细节
mosquitto_pub -h localhost -t "debug/test" -m "test" -d
# 输出:
# Client (sn-XXXX) sending CONNECT
# Client (sn-XXXX) received CONNACK (0)
# Client (sn-XXXX) sending PUBLISH (d0, q0, r0, m0, 'debug/test', ... (4 bytes))
# Client (sn-XXXX) received PUBACK (m0, rc=0)

mosquitto_sub

Basic Usage

# 基本语法
mosquitto_sub -h <host> -t <topic>

# 常用参数
-t, --topic     订阅 Topic（支持通配符）
-v, --verbose   显示 Topic 和消息
-C, --count     接收指定数量消息后退出
-T, --topic-filter 过滤显示的 Topic
-I, --id        指定客户端 ID
--retained-only 只接收保留消息

Practical Examples

# 1. 简单订阅
mosquitto_sub -h localhost -t "test/topic"

# 2. 显示 Topic 名称（-v）
mosquitto_sub -h localhost -t "sensors/#" -v

# 3. 接收指定数量后退出
mosquitto_sub -h localhost -t "test/topic" -C 5

# 4. 使用通配符
mosquitto_sub -h localhost -t "factory/+/temperature" -v

# 5. 指定 QoS
mosquitto_sub -h localhost -t "devices/control" -q 1

# 6. 只接收保留消息
mosquitto_sub -h localhost -t "#" --retained-only -v

# 7. 调试模式
mosquitto_sub -h localhost -t "test/#" -d -v

CLI Scripting

Shell Script Examples

#!/bin/bash
# mqtt_monitor.sh - 监控传感器数据

BROKER="localhost"
TOPIC="sensors/#"

echo "Starting MQTT Monitor..."
echo "Broker: $BROKER"
echo "Topic: $TOPIC"
echo "================================"

# 订阅并格式化输出
mosquitto_sub -h $BROKER -t $TOPIC -v | while read line; do
  TIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S')
  echo "[$TIMESTAMP] $line"
done

#!/bin/bash
# mqtt_publish_test.sh - 模拟传感器数据

BROKER="localhost"
INTERVAL=3  # 每 3 秒发布一次

echo "Publishing simulated sensor data..."
echo "Press Ctrl+C to stop"

while true; do
  TEMP=$(awk -v min=20 -v max=30 'BEGIN{srand(); print min+rand()*(max-min)}')
  HUM=$(awk -v min=40 -v max=80 'BEGIN{srand(); print min+rand()*(max-min)}')

  mosquitto_pub -h $BROKER \
    -t "sensors/environment" \
    -m "{\"temperature\": $TEMP, \"humidity\": $HUM}"

  printf "Published: temp=%.1f°C, hum=%.1f%%\n" $TEMP $HUM
  sleep $INTERVAL
done

Batch Testing

#!/bin/bash
# mqtt_bulk_test.sh - 批量测试

# 同时测试 10 个设备
for i in $(seq 1 10); do
  DEVICE_ID="esp32_test_$(printf "%03d" $i)"
  mosquitto_pub -h localhost \
    -t "devices/$DEVICE_ID/test" \
    -m "{\"test_id\": $i, \"message\": \"Test message from $DEVICE_ID\"}"
  echo "Sent message from $DEVICE_ID"
done

Mosquitto Passwd

User Management

# 创建密码文件
mosquitto_passwd -c /etc/mosquitto/passwd user1
# 提示输入密码

# 添加用户到现有密码文件
mosquitto_passwd -b /etc/mosquitto/passwd user2 password123

# 删除用户
mosquitto_passwd -D /etc/mosquitto/passwd user1

# 更新用户密码
mosquitto_passwd -b /etc/mosquitto/passwd user1 newpassword

# 查看密码文件
cat /etc/mosquitto/passwd
# user1:$6$salt$hash...

Mosquitto Sub/Sub in Docker

Docker Container Operations

# 进入 Mosquitto 容器
docker exec -it mosquitto sh

# 在容器内使用 CLI
docker exec mosquitto mosquitto_pub \
  -h localhost -t "docker/test" -m "Hello from Docker"

# 查看容器日志
docker logs mosquitto --tail 50

# 查看实时日志
docker logs -f mosquitto

$SYS Monitoring Topics

System Topics

Mosquitto 提供以 $SYS 开头的系统 Topic 用于监控：

# 订阅系统信息
mosquitto_sub -h localhost -t '$SYS/broker/#' -v

# 常用 $SYS Topics
$SYS/broker/version                    # Broker 版本
$SYS/broker/uptime                     # 运行时间
$SYS/broker/clients/total              # 客户端总数
$SYS/broker/clients/connected          # 当前连接数
$SYS/broker/clients/disconnected       # 断开连接数
$SYS/broker/messages/sent              # 发送消息总数
$SYS/broker/messages/received          # 接收消息总数
$SYS/broker/bytes/sent                 # 发送字节总数
$SYS/broker/bytes/received             # 接收字节总数
$SYS/broker/subscriptions/count        # 订阅总数
$SYS/broker/retained messages/count   # 保留消息数量
$SYS/broker/heap/current              # 当前堆内存使用
$SYS/broker/heap/maximum              # 最大堆内存使用

# 监控脚本示例
#!/bin/bash
# 每 10 秒输出 Broker 状态
while true; do
  echo "=== Broker Status ==="
  mosquitto_sub -h localhost -t '$SYS/broker/clients/connected' --retained-only -C 1
  mosquitto_sub -h localhost -t '$SYS/broker/messages/received' --retained-only -C 1
  mosquitto_sub -h localhost -t '$SYS/broker/uptime' --retained-only -C 1
  sleep 10
done

Troubleshooting CLI

Common CLI Issues

Issue 1: 连接被拒绝

# 错误
Error: Connection refused

# 排查
ping localhost                    # 检查网络
docker ps | grep mosquitto       # 检查 Broker 是否运行
netstat -an | grep 1883          # 检查端口

Issue 2: 认证失败

# 错误
Connection error: Connection refused (5)

# 检查
# 1. 确认用户名密码正确
# 2. 检查 mosquitto.conf
grep "allow_anonymous\|password_file" /etc/mosquitto/mosquitto.conf

# 临时允许匿名连接测试
# 修改配置为: allow_anonymous true

Issue 3: 通配符不工作

# 检查 Topic 层级
mosquitto_pub -h localhost -t "a/b/c" -m "test"
mosquitto_sub -h localhost -t "a/#" -v   # ✅ 工作
mosquitto_sub -h localhost -t "a/+" -v    # ✅ 工作
mosquitto_sub -h localhost -t "+/b/#" -v  # ✅ 工作
mosquitto_sub -h localhost -t "#" -v      # ✅ 全局订阅

Summary

本节要点总结：

1. mosquitto_pub：发布消息，支持 QoS、保留、调试模式
2. mosquitto_sub：订阅消息，支持通配符、verbose 输出、计数退出
3. 脚本应用：模拟数据生成、批量测试、监控脚本
4. 用户管理：mosquitto_passwd 创建和管理认证用户
5. $SYS Topics：通过系统 Topic 监控 Broker 运行状态
6. Docker 集成：在 Docker 环境中执行 MQTT CLI 命令