nats

NATS Core の知識

完全なリファレンス：JetStream のパターン（Node.js、Java）、セキュリティ設定、TLS のセットアップ、クラスタリングについては、advanced.md を参照してください。

詳細な知識：包括的なドキュメントを参照するには、technology: nats を指定して mcp__documentation__fetch_docs を使用してください。

クイックスタート（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
永続化	なし	あり
配信	最大 1 回	最低 1 回
リプレイ	なし	あり
コンシューマーグループ	キューグループ	コンシューマーグループ

サブジェクトのパターン

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