nats

NATS 核心知识

完整参考：JetStream 模式（Node.js、Java）、安全配置、TLS 设置和集群相关内容，请参阅 advanced.md。

深入了解：使用 mcp__documentation__fetch_docs 并将技术指定为 nats，获取全面的文档。

快速开始（Docker）

# docker-compose.yml
services:
  nats:
    image: nats:latest
    ports:
      - "4222:4222"   # Client
      - "8222:8222"   # Monitoring
    command: "-js -m 8222"  # Enable JetStream

核心概念

功能	Core NATS	JetStream
持久化	否	是
投递	至多一次	至少一次
重放	否	是
消费者组	队列组	消费者组

主题模式

orders.created     → Specific subject
orders.*           → orders.created, orders.updated (single wildcard)
orders.>           → orders.created, orders.us.east (multi wildcard)

Core NATS（Node.js）

import { connect, StringCodec } from 'nats';

const nc = await connect({ servers: 'localhost:4222' });
const sc = StringCodec();

// Simple publish
nc.publish('orders.created', sc.encode(JSON.stringify(order)));

// Subscribe
const sub = nc.subscribe('orders.*');
for await (const msg of sub) {
  const order = JSON.parse(sc.decode(msg.data));
  console.log(`Received on ${msg.subject}:`, order);
}

// Queue group (load balancing)
const qsub = nc.subscribe('orders.process', { queue: 'workers' });

// Request/Reply
const response = await nc.request('orders.validate', sc.encode(JSON.stringify(order)), {
  timeout: 5000,
});
const result = JSON.parse(sc.decode(response.data));

await nc.drain();

Core NATS（Python）

import nats
import json
import asyncio

async def main():
    nc = await nats.connect("nats://localhost:4222")

    # Publish
    await nc.publish("orders.created", json.dumps(order).encode())

    # Subscribe
    async def message_handler(msg):
        order = json.loads(msg.data.decode())
        print(f"Received: {order}")

    await nc.subscribe("orders.*", cb=message_handler)

    # Request/Reply
    response = await nc.request("orders.validate",
                                json.dumps(order).encode(),
                                timeout=5)
    result = json.loads(response.data.decode())

    await nc.drain()

asyncio.run(main())

Core NATS（Go）

nc, _ := nats.Connect(nats.DefaultURL)
defer nc.Drain()

// Publish
body, _ := json.Marshal(order)
nc.Publish("orders.created", body)

// Subscribe
nc.Subscribe("orders.*", func(msg *nats.Msg) {
    var order Order
    json.Unmarshal(msg.Data, &order)
    processOrder(order)
})

// Queue group
nc.QueueSubscribe("orders.process", "workers", func(msg *nats.Msg) {
    // Load balanced
})

// Request/Reply
response, _ := nc.Request("orders.validate", body, 5*time.Second)

不适用此技能的情况

• 需要长期事件重放——Kafka 提供更好的保留能力
• 路由模式复杂——RabbitMQ 交换器更加灵活
• 需要符合 JMS 标准——使用 ActiveMQ
• 需要与 AWS 原生集成——SQS 的集成效果更好

反模式

反模式	不妥之处	解决方案
使用 Core NATS 处理关键数据	即发即弃	使用 JetStream 实现持久化
没有主题层级	难以筛选	使用以点分隔的主题
请求没有超时	请求可能挂起	始终设置超时时间
载荷过大	增加网络负担	保持消息精简
在生产环境中使用单台服务	没有高可用性	部署包含 3 台或更多服务的集群

快速故障排查

问题	可能原因	解决方法
未收到消息	主题错误	验证主题名称并检查通配符
请求超时	没有响应方	验证响应方是否存在
JetStream 错误	未启用	使用 -js 标志启动服务
消费者延迟持续增长	处理速度较慢	增加消费者或优化处理
ACK 超时	处理速度过慢	增加 ack_wait

生产环境检查清单

• 已启用 TLS
• 已配置身份验证
• 已配置授权（权限）
• JetStream 副本数 >= 3
• 已配置流限制
• 已设置消费者 max_deliver
• 已配置 ACK 超时时间
• 已设置监控仪表板
• 已配置集群运行状况检查

参考文档

可用主题：basics、jetstream、patterns、production