突破连接边界：EMQX 实现 MQTT 和 NATS 协议双向互通

随着最新版本 EMQX 5.10.0 的发布，EMQX 协议网关家族又迎来了一位新成员：EMQX NATS 网关。这项新功能进一步扩展了 EMQX 的连接边界，实现了 MQTT 与 NATS 协议之间的原生、双向互通，为构建更灵活、更强大的实时数据基础设施提供了前所未有的可能性。

docker run --name emqx \
 -p 18083:18083 -p 1883:1883 -p 20243:20243 \
 -d emqx/emqx-enterprise:5.10.0

import asyncio
import nats
async def run():
    nc = await nats.connect(servers=["nats://localhost:20243"])
    print("NATS Publisher connected to EMQX NATS Gateway.")
    subject = "sensor.data.temperature"
    message = b'{"device_id": "sensor_001", "temp": 25.5}'
    await nc.publish(subject, message)
    print(f"Published NATS message to subject '{subject}': {message.decode()}")
    await nc.drain()
    print("NATS Publisher disconnected.")
if __name__ == '__main__':
    asyncio.run(run())

import paho.mqtt.client as paho
from paho import mqtt
import time
# MQTT 消息回调函数
def on_message(client, userdata, msg):
    print(f"Received MQTT message on topic '{msg.topic}': {msg.payload.decode()}")
def run():
    client = paho.Client(client_id="", userdata=None, protocol=paho.MQTTv5)
    client.on_message = on_message
    # 连接到 EMQX MQTT 监听器 (默认端口 1883)
    client.connect("localhost", 1883, 60)
    print("MQTT Subscriber connected to EMQX.")
    # 订阅映射后的 MQTT 主题
    # 根据 NATS Gateway 的 topic_mapping 规则，iot.sensor.data.temperature 映射到 sensor/data/temperature
    client.subscribe("sensor/data/temperature", qos=1)
    print("MQTT Subscriber subscribed to 'sensor/data/temperature'.")
    client.loop_forever()
if __name__ == '__main__':
    run()

import paho.mqtt.client as paho
from paho import mqtt
import time
def run():
    client = paho.Client(client_id="", userdata=None, protocol=paho.MQTTv5)
    # 连接到 EMQX MQTT 监听器 (默认端口 1883)
    client.connect("localhost", 1883, 60)
    print("MQTT Publisher connected to EMQX.")
    topic = "command/device/light_001"
    message = '{"action": "turn_on", "brightness": 80}'
    client.publish(topic, message, qos=1)
    print(f"Published MQTT message to topic '{topic}': {message}")
    client.disconnect()
    print("MQTT Publisher disconnected.")
if __name__ == '__main__':
    run()

import asyncio
import nats
async def message_handler(msg):
    print(f"Received NATS message on subject '{msg.subject}': {msg.data.decode()}")
async def run():
    # 连接到 EMQX NATS Gateway
    nc = await nats.connect(servers=["nats://localhost:20243"])
    print("NATS Subscriber connected to EMQX NATS Gateway.")
    # 订阅映射后的 NATS Subject
    # command/device/light_001 映射到 command.device.light_001
    await nc.subscribe("command.device.light_001", cb=message_handler)
    print("NATS Subscriber subscribed to 'device.command.light_001'.")
    # 保持连接，等待消息
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        pass
    finally:
        await nc.drain()
        print("NATS Subscriber disconnected.")
if __name__ == '__main__':
    asyncio.run(run())