MQTT 长连接通信

import paho.mqtt.client as mqtt
import json, time

# ---- 回调函数 ----
def on_connect(client, userdata, flags, rc):
    print(f"Connected: rc={rc}")
    # 连接成功后订阅
    client.subscribe("sensor/#", qos=1)

def on_message(client, userdata, msg):
    payload = json.loads(msg.payload.decode())
    print(f"[{msg.topic}] {payload}")

def on_disconnect(client, userdata, rc):
    print(f"Disconnected: rc={rc}")
    if rc != 0:
        client.reconnect()  # 自动重连

# ---- 客户端配置 ----
client = mqtt.Client(client_id="device_001", clean_session=False)
client.username_pw_set("user", "pass")

# 遗嘱消息
client.will_set(
    topic="devices/device_001/status",
    payload=json.dumps({"status": "offline"}),
    qos=1, retain=True
)

client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect

# 建立长连接（keepalive=60s）
client.connect("broker.emqx.io", port=1883, keepalive=60)
client.loop_start()  # 后台线程维持连接

# 发布消息
while True:
    msg = {"temp": 25.6, "ts": int(time.time())}
    client.publish("sensor/temp/room1", json.dumps(msg), qos=1)
    time.sleep(5)

import mqtt from 'mqtt'

// WebSocket 连接（浏览器端用 wss://）
const client = mqtt.connect('mqtt://broker.emqx.io:1883', {
  clientId: `device_${Math.random().toString(16).slice(2)}`,
  username: 'user',
  password: 'pass',
  keepalive: 60,
  clean: false,
  reconnectPeriod: 3000,  // 断线 3s 后自动重连
  will: {
    topic: 'devices/status',
    payload: JSON.stringify({ status: 'offline' }),
    qos: 1,
    retain: true
  }
})

client.on('connect', () => {
  console.log('Connected!')
  // 订阅（支持多 topic）
  client.subscribe(['sensor/#', 'camera/+'], { qos: 1 })
  // 发布上线状态
  client.publish('devices/status', JSON.stringify({ status: 'online' }), {
    qos: 1, retain: true
  })
})

client.on('message', (topic, payload) => {
  const msg = JSON.parse(payload.toString())
  console.log(`[${topic}]`, msg)
})

client.on('reconnect', () => console.log('Reconnecting...'))
client.on('error', err => console.error('MQTT Error:', err))

package main

import (
    "encoding/json"
    "fmt"
    "time"
    mqtt "github.com/eclipse/paho.mqtt.golang"
)

func main() {
    opts := mqtt.NewClientOptions()
    opts.AddBroker("tcp://broker.emqx.io:1883")
    opts.SetClientID("go_device_001")
    opts.SetKeepAlive(60 * time.Second)
    opts.SetCleanSession(false)

    // 自动重连
    opts.SetAutoReconnect(true)
    opts.SetMaxReconnectInterval(30 * time.Second)

    // 连接回调
    opts.SetOnConnectHandler(func(c mqtt.Client) {
        fmt.Println("Connected!")
        c.Subscribe("sensor/#", 1, func(c mqtt.Client, m mqtt.Message) {
            fmt.Printf("[%s] %s\n", m.Topic(), m.Payload())
        })
    })

    client := mqtt.NewClient(opts)
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        panic(token.Error())
    }

    // 周期发布消息
    for {
        data, _ := json.Marshal(map[string]interface{}{
            "temp": 25.6, "ts": time.Now().Unix(),
        })
        client.Publish("sensor/temp/room1", 1, false, data)
        time.Sleep(5 * time.Second)
    }
}