PUB <subject> [reply-to] <bytes>\r\n[payload]\r\n
SUB <subject> [queue group] <sid>\r\n
UNSUB <sid> [max_msgs]\r\n
MSG <subject> <sid> [reply-to] <bytes>\r\n[payload]\r\n

services.users.created
services.users.updated
services.products.created

docker run -p 4222:4222 -p 8222:8222 -p 6222:6222 --name nats-server nats:2.10.5-alpine

# Linux/macOS
curl -L https://github.com/nats-io/nats-server/releases/download/v2.10.5/nats-server-v2.10.5-linux-amd64.tar.gz -o nats-server.tar.gz
tar -xzf nats-server.tar.gz
cd nats-server-v2.10.5-linux-amd64
sudo cp nats-server /usr/local/bin/

# Windows (PowerShell)
Invoke-WebRequest -Uri https://github.com/nats-io/nats-server/releases/download/v2.10.5/nats-server-v2.10.5-windows-amd64.zip -OutFile nats-server.zip
Expand-Archive -Path nats-server.zip -DestinationPath C:\nats-server
$env:PATH += ";C:\nats-server\nats-server-v2.10.5-windows-amd64"

# server.conf
port: 4222
http_port: 8222

# 기본 로깅 설정
debug: false
trace: false
logtime: true

# 최대 연결 수
max_connections: 65536

# 최대 제어 라인 크기
max_control_line: 4096

# 최대 페이로드 크기 (기본값: 1MB)
max_payload: 1048576

# 핑 간격 (초 단위)
ping_interval: 120

# 최대 핑 아웃 수
ping_max: 2

nats-server -c server.conf

docker run -p 4222:4222 -p 8222:8222 -v $(pwd)/server.conf:/server.conf nats:2.10.5-alpine -c /server.conf

curl http://localhost:8222/varz

{
  "server_id": "NCDKYTJKBQXTW6QOQB5IOAQTJG3XSVQHQCFGMQPQBZB3VWCWSEVLZUID",
  "server_name": "NCDKYTJKBQXTW6QOQB5IOAQTJG3XSVQHQCFGMQPQBZB3VWCWSEVLZUID",
  "version": "2.10.5",
  "proto": 1,
  "go": "go1.21.0",
  "host": "0.0.0.0",
  "port": 4222,
  "max_connections": 65536,
  "ping_interval": 120000000000,
  "ping_max": 2,
  "http_host": "0.0.0.0",
  "http_port": 8222,
  "https_port": 0,
  "auth_timeout": 2000000000,
  "max_control_line": 4096,
  "max_payload": 1048576,
  "max_pending": 67108864,
  "cluster": {},
  "gateway": {},
  "leaf": {},
  "websocket": {},
  "mqtt": {},
  "jetstream": {},
  "tls_timeout": 0.5,
  "write_deadline": 10000000000,
  "start": "2025-04-02T10:00:00.123456789Z",
  "now": "2025-04-02T10:05:00.123456789Z",
  "uptime": "5m0s",
  "mem": 12582912,
  "cores": 8,
  "cpu": 0.5,
  "connections": 0,
  "total_connections": 0,
  "routes": 0,
  "remotes": 0,
  "leafnodes": 0,
  "in_msgs": 0,
  "out_msgs": 0,
  "in_bytes": 0,
  "out_bytes": 0,
  "slow_consumers": 0,
  "subscriptions": 0,
  "config_load_time": "2025-04-02T10:00:00.123456789Z"
}

# Linux/macOS
curl -L https://github.com/nats-io/natscli/releases/download/v0.1.1/nats-0.1.1-linux-amd64.zip -o nats-cli.zip
unzip nats-cli.zip
sudo cp nats-0.1.1-linux-amd64/nats /usr/local/bin/

# Windows (PowerShell)
Invoke-WebRequest -Uri https://github.com/nats-io/natscli/releases/download/v0.1.1/nats-0.1.1-windows-amd64.zip -OutFile nats-cli.zip
Expand-Archive -Path nats-cli.zip -DestinationPath C:\nats-cli
$env:PATH += ";C:\nats-cli\nats-0.1.1-windows-amd64"

# 터미널 1에서 구독자 실행
nats sub "test.subject"

# 터미널 2에서 발행자 실행
nats pub "test.subject" "Hello NATS World!"

// 설치
// go get github.com/nats-io/nats.go

package main

import (
    "fmt"
    "log"
    "time"

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

func main() {
    // NATS 서버에 연결
    nc, err := nats.Connect("nats://localhost:4222")
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Close()
    
    // 구독 설정
    sub, err := nc.Subscribe("greetings", func(msg *nats.Msg) {
        fmt.Printf("메시지 받음: %s\n", string(msg.Data))
        // 응답 보내기
        nc.Publish(msg.Reply, []byte("안녕하세요! 반갑습니다!"))
    })
    if err != nil {
        log.Fatal(err)
    }
    defer sub.Unsubscribe()
    
    // 요청-응답 패턴 사용
    msg, err := nc.Request("greetings", []byte("안녕하세요!"), 2*time.Second)
    if err != nil {
        log.Fatal(err)
    }
    
    fmt.Printf("응답 받음: %s\n", string(msg.Data))
    
    // 잠시 대기
    time.Sleep(1 * time.Second)
}

// 설치
// npm install nats

const { connect } = require('nats');

async function main() {
    try {
        // NATS 서버에 연결
        const nc = await connect({ servers: 'nats://localhost:4222' });
        
        console.log('NATS 서버에 연결됨!');
        
        // 구독 설정
        const sub = nc.subscribe('orders.new', {
            callback: (err, msg) => {
                if (err) {
                    console.error('구독 에러:', err);
                    return;
                }
                
                console.log(`주문 받음: ${msg.data}`);
                
                // 주문 처리 로직...
                
                // 응답 보내기 (만약 reply 주제가 있다면)
                if (msg.reply) {
                    nc.publish(msg.reply, new TextEncoder().encode('주문이 처리되었습니다!'));
                }
            }
        });
        
        // 메시지 발행
        nc.publish('orders.new', new TextEncoder().encode(JSON.stringify({
            id: 'order-123',
            product: '맛있는 케이크',
            quantity: 2,
            customer: '홍길동'
        })));
        
        // 요청-응답 패턴
        const response = await nc.request('orders.status', new TextEncoder().encode('order-123'), { timeout: 2000 });
        console.log('응답:', new TextDecoder().decode(response.data));
        
        // 잠시 후 연결 종료
        setTimeout(async () => {
            await sub.unsubscribe();
            await nc.close();
            console.log('NATS 연결 종료!');
        }, 5000);
        
    } catch (error) {
        console.error('에러 발생:', error);
    }
}

main();

# 설치
# pip install nats-py

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

async def main():
    # NATS 클라이언트 생성
    nc = NATS()
    
    # NATS 서버에 연결
    await nc.connect("nats://localhost:4222")
    print("NATS 서버에 연결됨!")
    
    # 구독 콜백 함수
    async def message_handler(msg):
        subject = msg.subject
        data = msg.data.decode()
        print(f"메시지 받음 [{subject}]: {data}")
        
        # 응답 보내기 (만약 reply가 있다면)
        if msg.reply:
            await nc.publish(msg.reply, b"메시지 잘 받았습니다!")
    
    # 주제 구독
    await nc.subscribe("updates.>", cb=message_handler)
    
    # 메시지 발행
    await nc.publish("updates.user", json.dumps({
        "id": 123,
        "name": "김철수",
        "action": "login"
    }).encode())
    
    # 요청-응답 패턴
    try:
        response = await nc.request("service.time", b"", timeout=1)
        print(f"현재 시간: {response.data.decode()}")
    except Exception as e:
        print(f"요청 에러: {e}")
    
    # 잠시 대기 (메시지 처리 시간)
    await asyncio.sleep(2)
    
    # 연결 종료
    await nc.close()
    print("NATS 연결 종료!")

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

// 설치 (Maven)
// <dependency>
//     <groupId>io.nats</groupId>
//     <artifactId>jnats</artifactId>
//     <version>2.17.1</version>
// </dependency>

import io.nats.client.*;
import io.nats.client.api.*;

import java.nio.charset.StandardCharsets;
import java.time.Duration;

public class NatsExample {
    public static void main(String[] args) {
        try {
            // NATS 서버에 연결
            Connection nc = Nats.connect("nats://localhost:4222");
            System.out.println("NATS 서버에 연결됨!");
            
            // 구독 설정
            Dispatcher dispatcher = nc.createDispatcher((msg) -> {
                String subject = msg.getSubject();
                String data = new String(msg.getData(), StandardCharsets.UTF_8);
                System.out.printf("메시지 받음 [%s]: %s\n", subject, data);
                
                // 응답 보내기 (만약 reply가 있다면)
                if (msg.getReplyTo() != null) {
                    nc.publish(msg.getReplyTo(), "메시지 잘 받았습니다!".getBytes(StandardCharsets.UTF_8));
                }
            });
            
            // 주제 구독
            dispatcher.subscribe("notifications.>");
            
            // 메시지 발행
            nc.publish("notifications.email", 
                       String.format("{\"to\":\"user@example.com\",\"subject\":\"안녕하세요\",\"body\":\"NATS 테스트 메일입니다.\"}").getBytes(StandardCharsets.UTF_8));
            
            // 요청-응답 패턴
            Message response = nc.request("service.status", "check".getBytes(StandardCharsets.UTF_8), Duration.ofSeconds(2));
            System.out.println("응답: " + new String(response.getData(), StandardCharsets.UTF_8));
            
            // 잠시 대기
            Thread.sleep(1000);
            
            // 연결 종료
            nc.close();
            System.out.println("NATS 연결 종료!");
            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

# JetStream 활성화하여 NATS 서버 실행
nats-server -js

# 또는 Docker로
docker run -p 4222:4222 -p 8222:8222 nats:2.10.5-alpine -js

package main

import (
    "fmt"
    "log"
    "time"

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

func main() {
    // NATS 서버에 연결
    nc, err := nats.Connect("nats://localhost:4222")
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Close()
    
    // JetStream 컨텍스트 생성
    js, err := nc.JetStream()
    if err != nil {
        log.Fatal(err)
    }
    
    // 스트림 생성 (아직 없는 경우)
    streamConfig := &nats.StreamConfig{
        Name:     "ORDERS",
        Subjects: []string{"orders.*"},
        Storage:  nats.FileStorage,
        MaxAge:   24 * time.Hour, // 메시지 보관 기간
    }
    
    _, err = js.AddStream(streamConfig)
    if err != nil {
        log.Fatal(err)
    }
    
    // 메시지 발행 (지속성 있음)
    _, err = js.Publish("orders.new", []byte("새로운 주문: 아이스크림 3개"))
    if err != nil {
        log.Fatal(err)
    }
    
    // 컨슈머 생성 및 구독
    sub, err := js.PullSubscribe("orders.*", "order-processor")
    if err != nil {
        log.Fatal(err)
    }
    
    // 메시지 가져오기
    msgs, err := sub.Fetch(10)
    if err != nil {
        log.Fatal(err)
    }
    
    for _, msg := range msgs {
        fmt.Printf("메시지 받음: %s\n", string(msg.Data))
        msg.Ack() // 메시지 처리 완료 확인
    }
}

// Go 예제
// 구독자 1
nc.QueueSubscribe("tasks.process", "workers", func(msg *nats.Msg) {
    fmt.Printf("구독자 1: 작업 처리 중 - %s\n", string(msg.Data))
    // 작업 처리 로직...
})

// 구독자 2
nc.QueueSubscribe("tasks.process", "workers", func(msg *nats.Msg) {
    fmt.Printf("구독자 2: 작업 처리 중 - %s\n", string(msg.Data))
    // 작업 처리 로직...
})

// 구독자 3
nc.QueueSubscribe("tasks.process", "workers", func(msg *nats.Msg) {
    fmt.Printf("구독자 3: 작업 처리 중 - %s\n", string(msg.Data))
    // 작업 처리 로직...
})

// 서비스 제공자 (Node.js)
nc.subscribe('services.math.add', (err, msg) => {
    if (err) return;
    
    const data = JSON.parse(new TextDecoder().decode(msg.data));
    const result = data.a + data.b;
    
    nc.publish(msg.reply, new TextEncoder().encode(JSON.stringify({ result })));
});

// 서비스 소비자 (Node.js)
async function addNumbers(a, b) {
    const response = await nc.request(
        'services.math.add',
        new TextEncoder().encode(JSON.stringify({ a, b })),
        { timeout: 1000 }
    );
    
    return JSON.parse(new TextDecoder().decode(response.data)).result;
}

// 사용 예
const sum = await addNumbers(5, 3);
console.log(`5 + 3 = ${sum}`); // "5 + 3 = 8"

// Go 예제
// 요청자
inbox := nc.NewInbox()
sub, _ := nc.SubscribeSync(inbox)

// 모든 서비스에 요청 브로드캐스트
nc.PublishRequest("services.status", inbox, []byte("status check"))

// 응답 수집 (타임아웃 설정)
responses := []string{}
timeout := time.After(500 * time.Millisecond)

for {
    select {
    case <-timeout:
        // 타임아웃 발생, 응답 수집 종료
        fmt.Printf("%d개 서비스에서 응답 받음\n", len(responses))
        for i, resp := range responses {
            fmt.Printf("응답 %d: %s\n", i+1, resp)
        }
        return
    default:
        // 응답 대기
        msg, err := sub.NextMsg(10 * time.Millisecond)
        if err == nil {
            responses = append(responses, string(msg.Data))
        }
    }
}

// 주제 계층 구조 예시
events.user.created
events.user.updated
events.user.deleted
events.product.created
events.product.updated
events.product.deleted
events.order.created
events.order.completed

// 와일드카드 구독 예시
nc.Subscribe("events.user.>", handler)  // 모든 사용자 이벤트
nc.Subscribe("events.*.created", handler)  // 모든 생성 이벤트
nc.Subscribe("events.>", handler)  // 모든 이벤트

# 서버 1 설정 (server1.conf)
port: 4222
http_port: 8222
cluster {
  port: 6222
  routes: [
    nats-route://localhost:6223
    nats-route://localhost:6224
  ]
}

# 서버 2 설정 (server2.conf)
port: 4223
http_port: 8223
cluster {
  port: 6223
  routes: [
    nats-route://localhost:6222
    nats-route://localhost:6224
  ]
}

# 서버 3 설정 (server3.conf)
port: 4224
http_port: 8224
cluster {
  port: 6224
  routes: [
    nats-route://localhost:6222
    nats-route://localhost:6223
  ]
}

# 터미널 1
nats-server -c server1.conf

# 터미널 2
nats-server -c server2.conf

# 터미널 3
nats-server -c server3.conf

# docker-compose.yml
version: '3'
services:
  nats1:
    image: nats:2.10.5-alpine
    command: "--cluster nats://0.0.0.0:6222 --cluster_name NATS --cluster_routes nats://nats2:6222,nats://nats3:6222 --http_port 8222"
    ports:
      - "4222:4222"
      - "8222:8222"
  
  nats2:
    image: nats:2.10.5-alpine
    command: "--cluster nats://0.0.0.0:6222 --cluster_name NATS --cluster_routes nats://nats1:6222,nats://nats3:6222 --http_port 8222"
    ports:
      - "4223:4222"
      - "8223:8222"
  
  nats3:
    image: nats:2.10.5-alpine
    command: "--cluster nats://0.0.0.0:6222 --cluster_name NATS --cluster_routes nats://nats1:6222,nats://nats2:6222 --http_port 8222"
    ports:
      - "4224:4222"
      - "8224:8222"

// Go 예제
servers := []string{
    "nats://localhost:4222",
    "nats://localhost:4223",
    "nats://localhost:4224",
}

// 연결 옵션 설정
opts := nats.Options{
    Servers: servers,
    MaxReconnects: -1,  // 무제한 재연결 시도
    ReconnectWait: 2 * time.Second,
    DisconnectErrHandler: func(nc *nats.Conn, err error) {
        fmt.Printf("연결 끊김: %v\n", err)
    },
    ReconnectHandler: func(nc *nats.Conn) {
        fmt.Printf("서버 %s에 재연결됨\n", nc.ConnectedUrl())
    },
}

// 연결
nc, err := opts.Connect()
if err != nil {
    log.Fatal(err)
}

# 데이터센터 A의 서버 설정
gateway {
  name: "A"
  port: 7222
  gateways: [
    {name: "B", url: "nats://dc-b-gateway:7222"}
    {name: "C", url: "nats://dc-c-gateway:7222"}
  ]
}

# 데이터센터 B의 서버 설정
gateway {
  name: "B"
  port: 7222
  gateways: [
    {name: "A", url: "nats://dc-a-gateway:7222"}
    {name: "C", url: "nats://dc-c-gateway:7222"}
  ]
}

// Go 예제
streamConfig := &nats.StreamConfig{
    Name:     "ORDERS",
    Subjects: []string{"orders.*"},
    Storage:  nats.FileStorage,
    Replicas: 3,  // 3개 서버에 복제
}

# 클러스터 정보 확인
curl http://localhost:8222/varz

# 연결된 클라이언트 확인
curl http://localhost:8222/connz

# 구독 정보 확인
curl http://localhost:8222/subsz

# 라우트 정보 확인
curl http://localhost:8222/routez

# 주요 모니터링 엔드포인트
http://localhost:8222/varz    # 서버 일반 정보
http://localhost:8222/connz   # 연결 정보
http://localhost:8222/routez  # 라우트 정보
http://localhost:8222/subsz   # 구독 정보
http://localhost:8222/jsz     # JetStream 정보

NATS 모니터링 대시보드 예시

# prometheus.yml
scrape_configs:
  - job_name: 'nats'
    static_configs:
      - targets: ['nats1:8222', 'nats2:8222', 'nats3:8222']
    metrics_path: /metrics

nats-server --http_port=8222 --prometheus

# 느린 소비자 알림 예시 (PromQL)
sum(nats_slow_consumers) > 0

# 높은 메모리 사용량 알림
nats_mem > 1024 * 1024 * 1024  # 1GB 이상

# 연결 수 급증 알림
rate(nats_connections[5m]) > 100

# server.conf
debug: false   # 디버그 로그 (개발 환경에서만 권장)
trace: false   # 상세 추적 로그 (문제 해결 시에만 사용)
logtime: true  # 로그에 타임스탬프 포함
log_file: "/var/log/nats-server.log"  # 로그 파일 경로

# server.conf
max_payload: 2097152  # 2MB로 증가
max_pending: 67108864  # 64MB로 설정

# server.conf
max_connections: 10000

# server.conf
ping_interval: 60  # 60초마다 핑
ping_max: 2  # 2번 핑 실패 시 연결 종료

# server.conf
authorization {
  user: admin
  password: s3cr3t!
}

# 또는 여러 사용자 설정
authorization {
  users = [
    {user: admin, password: admin_password}
    {user: client, password: client_password}
    {user: readonly, password: readonly_password}
  ]
}

# server.conf
authorization {
  token: "s3cr3t_t0k3n"
}

# server.conf
tls {
  cert_file: "/path/to/server-cert.pem"
  key_file: "/path/to/server-key.pem"
  ca_file: "/path/to/ca-cert.pem"
  verify: true
}

# server.conf
authorization {
  users = [
    { nkey: UAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA }
    { nkey: UBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB }
  ]
}

# server.conf
operator: /path/to/operator.jwt
resolver: {
  type: full
  dir: /path/to/jwt/dir
}

# server.conf
authorization {
  users = [
    {
      user: admin
      password: admin_password
      permissions: {
        publish: ">"      # 모든 주제에 발행 가능
        subscribe: ">"    # 모든 주제 구독 가능
      }
    },
    {
      user: frontend
      password: frontend_password
      permissions: {
        publish: ["api.request.*", "logs.frontend.*"]
        subscribe: ["api.response.*", "notifications.>"]
      }
    },
    {
      user: readonly
      password: readonly_password
      permissions: {
        publish: {}       # 발행 권한 없음
        subscribe: "logs.>" # 로그 관련 주제만 구독 가능
      }
    }
  ]
}

# server.conf
tls {
  cert_file: "/path/to/server-cert.pem"
  key_file: "/path/to/server-key.pem"
  ca_file: "/path/to/ca-cert.pem"  # 클라이언트 인증 필요시
  verify: true                      # 클라이언트 인증서 검증
  cipher_suites: [                  # 사용할 암호화 알고리즘
    "TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384",
    "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"
  ]
  curve_preferences: [              # 사용할 타원 곡선
    "CurveP521",
    "CurveP384"
  ]
  timeout: 2                        # TLS 핸드셰이크 타임아웃(초)
}

// Go 예제
cert, _ := tls.LoadX509KeyPair("client-cert.pem", "client-key.pem")
caCert, _ := os.ReadFile("ca-cert.pem")
caCertPool := x509.NewCertPool()
caCertPool.AppendCertsFromPEM(caCert)

tlsConfig := &tls.Config{
    Certificates: []tls.Certificate{cert},
    RootCAs:      caCertPool,
    MinVersion:   tls.VersionTLS12,
}

nc, err := nats.Connect("tls://nats-server:4222",
    nats.Secure(tlsConfig))
if err != nil {
    log.Fatal(err)
}

# server.conf
# 특정 IP에서만 접속 허용
net: 192.168.1.0/24

# 또는 여러 네트워크 지정
net: [
  192.168.1.0/24,
  10.0.0.0/8
]

# HTTP 모니터링 인터페이스 제한
http: 127.0.0.1

# iptables 예시 (Linux)
iptables -A INPUT -p tcp --dport 4222 -s 192.168.1.0/24 -j ACCEPT
iptables -A INPUT -p tcp --dport 4222 -j DROP

# 모니터링 포트는 내부망에서만 접근 가능하도록
iptables -A INPUT -p tcp --dport 8222 -s 127.0.0.1 -j ACCEPT
iptables -A INPUT -p tcp --dport 8222 -j DROP

# Dockerfile 예시
FROM nats:2.10.5-alpine

# 비루트 사용자로 실행
USER 1000:1000

# 설정 파일 복사
COPY server.conf /server.conf

# 필요한 포트만 노출
EXPOSE 4222 8222

# 보안 설정으로 실행
ENTRYPOINT ["nats-server", "-c", "/server.conf"]

# docker-compose.yml 예시
version: '3'
services:
  nats:
    build: .
    ports:
      - "4222:4222"
    volumes:
      - ./certs:/certs
      - ./data:/data
    environment:
      - NATS_PASSWORD_FILE=/run/secrets/nats_password
    secrets:
      - nats_password
    security_opt:
      - no-new-privileges:true
    read_only: true
    tmpfs:
      - /tmp
    restart: unless-stopped

secrets:
  nats_password:
    file: ./secrets/nats_password.txt