NATS 通讯协议及
nats-server
深度调研报告

package main

import (
    "fmt"
    "log"
    "time"

    "github.com/nats-io/nats.go"
)

func main() {
    // 连接到 NATS 服务器
    nc, err := nats.Connect(nats.DefaultURL)
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Close()

    // 发布-订阅示例
    // 订阅主题 "updates"
    sub, err := nc.Subscribe("updates", func(msg *nats.Msg) {
        fmt.Printf("Received a message on subject %s: %s\n", msg.Subject, string(msg.Data))
    })
    if err != nil {
        log.Fatal(err)
    }
    defer sub.Unsubscribe()

    // 发布消息到主题 "updates"
    err = nc.Publish("updates", []byte("Hello, NATS!"))
    if err != nil {
        log.Fatal(err)
    }

    // 请求-回复示例
    // 订阅服务请求主题 "service.request"
    _, err = nc.Subscribe("service.request", func(msg *nats.Msg) {
        fmt.Printf("Received a request: %s\n", string(msg.Data))
        // 处理请求并发送回复
        nc.Publish(msg.Reply, []byte("Response to: "+string(msg.Data)))
    })
    
    // 发送请求并等待回复
    response, err := nc.Request("service.request", []byte("Request data"), 2*time.Second)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Received a reply: %s\n", string(response.Data))
}

import { connect, StringCodec } from 'nats';

// 连接到 NATS 服务器
const nc = await connect({ servers: 'nats://localhost:4222' });
const sc = StringCodec();

// 模拟一个实时数据源，例如股票行情或体育赛事比分
setInterval(() => {
  const message = `Current update: ${new Date().toLocaleTimeString()}`;
  // 发布实时更新到主题 "live.updates"
  nc.publish("live.updates", sc.encode(message));
  console.log(`Published: ${message}`);
}, 1000); // 每秒推送一次

// 客户端订阅示例
const sub = nc.subscribe("live.updates", {
  callback: (err, msg) => {
    if (err) {
      console.error(err);
      return;
    }
    console.log(`[Subscriber] Received: ${sc.decode(msg.data)}`);
  },
});

import asyncio
from nats.aio.client import Client as NATS

async def iot_device(device_id):
    nc = NATS()
    await nc.connect(servers=["nats://localhost:4222"])

    # 模拟设备定期发送传感器数据
    while True:
        sensor_data = f"Device {device_id}: Sensor reading {asyncio.get_event_loop().time()}"
        print(f"Sending: {sensor_data}")
        await nc.publish(f"iot.data.{device_id}", sensor_data.encode())
        await asyncio.sleep(2)  # 每2秒发送一次数据

async def cloud_service():
    nc = NATS()
    await nc.connect(servers=["nats://localhost:4222"])

    async def message_handler(msg):
        subject = msg.subject
        data = msg.data.decode()
        print(f"Received on [{subject}]: {data}")
        # 在这里处理接收到的 IoT 数据

    # 订阅所有 IoT 设备的数据主题 "iot.data.>"
    await nc.subscribe("iot.data.>", cb=message_handler)
    print("Cloud service subscribed to iot.data.>")

async def main():
    # 启动模拟的 IoT 设备
    device_task = asyncio.create_task(iot_device("sensor01"))
    # 启动云端服务
    cloud_task = asyncio.create_task(cloud_service())
    await asyncio.gather(device_task, cloud_task)

if __name__ == '__main__':
    asyncio.run(main())