Instalación del Sistema de Mensajería NATS

NATS is a high-performance, lightweight messaging system designed for cloud-native applications. It provides publish-subscribe, request-reply, and distributed queue patterns with minimal overhead and excellent latency characteristics. This guide covers NATS installation, configuration, core messaging patterns, security, and clustering.

Tabla de Contenidos

• Prerequisites
• Installing NATS Server
• Basic Configuration
• Pub-Sub Messaging Pattern
• Request-Reply Pattern
• JetStream Persistent Storage
• NATS Clustering
• Security and Authentication
• Client Libraries
• Monitoring
• Conclusion

Requisitos Previos

Before installing NATS, ensure you have:

• Linux system (Ubuntu 20.04+, CentOS 8+, Debian 11+)
• Root or sudo access
• At least 1GB RAM available
• Internet connectivity for package downloads
• Basic understanding of messaging patterns

Installing NATS Server

NATS server is available as a single binary, making installation straightforward. Download the latest version:

cd /tmp
curl -L https://github.com/nats-io/nats-server/releases/download/v2.10.0/nats-server-v2.10.0-linux-amd64.zip -o nats-server.zip
unzip nats-server.zip
sudo mv nats-server /usr/local/bin/

Verify the installation:

nats-server --version

Create a dedicated user for NATS:

sudo useradd -r -s /bin/false nats

Create necessary directories:

sudo mkdir -p /etc/nats /var/lib/nats /var/log/nats
sudo chown -R nats:nats /var/lib/nats /var/log/nats

Create a systemd service file:

sudo tee /etc/systemd/system/nats.service <<EOF
[Unit]
Description=NATS Server
Documentation=https://nats.io
After=network.target
Requires=network.target

[Service]
Type=simple
User=nats
Group=nats
ExecStart=/usr/local/bin/nats-server -c /etc/nats/nats.conf
ExecReload=/bin/kill -HUP \$MAINPID
Restart=always
RestartSec=5
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target
EOF

Enable and start the service:

sudo systemctl daemon-reload
sudo systemctl enable nats
sudo systemctl start nats

Verify NATS is running:

sudo systemctl status nats
nats-top

Configuración Básica

Create a basic NATS configuration file at /etc/nats/nats.conf:

sudo tee /etc/nats/nats.conf <<EOF
# NATS Server Configuration

# Network and port settings
host: 0.0.0.0
port: 4222

# HTTP monitoring endpoint
http: 8222
http_base_path: /

# TLS configuration (optional)
# tls: {
#   cert_file: /etc/nats/certs/server.crt
#   key_file: /etc/nats/certs/server.key
# }

# Authentication
authorization {
  users: [
    {user: "default", password: "securepass123"}
    {user: "monitor", password: "monitorpass456"}
  ]
}

# JetStream storage configuration
jetstream {
  store_dir: /var/lib/nats
  max_memory: 512M
  max_file: 10GB
}

# Logging
log_file: /var/log/nats/nats.log
log_level: info

# Maximum number of connections
max_connections: 65536

# Default authentication timeout
auth_timeout: 3

# Ping/pong settings
ping_interval: 2m
ping_max: 2
EOF

Restart NATS to apply the configuration:

sudo systemctl restart nats

Pub-Sub Messaging Pattern

NATS pub-sub provides one-to-many message distribution. Multiple subscribers can listen on the same subject. Install the NATS CLI for testing:

cd /tmp
curl -L https://github.com/nats-io/natscli/releases/download/v0.1.0/nats-v0.1.0-linux-amd64.zip -o nats-cli.zip
unzip nats-cli.zip
sudo mv nats /usr/local/bin/

Test pub-sub by creating a subscriber:

nats sub "orders.created" --server nats://default:securepass123@localhost:4222

In another terminal, publish a message:

nats pub "orders.created" "Order #12345 created" --server nats://default:securepass123@localhost:4222

The subscriber receives the message. Create a wildcard subscriber to listen to multiple subjects:

nats sub "orders.>" --server nats://default:securepass123@localhost:4222

Publish to multiple subjects:

nats pub "orders.created" "Order #12345 created" --server nats://default:securepass123@localhost:4222
nats pub "orders.shipped" "Order #12345 shipped" --server nats://default:securepass123@localhost:4222

Create a Python client for pub-sub:

pip3 install asyncio-nats-client

Create a subscriber script (nats_subscriber.py):

#!/usr/bin/env python3
import asyncio
from nats.aio.client import Client as NATS

async def main():
    nc = NATS()
    await nc.connect(servers=["nats://default:securepass123@localhost:4222"])
    
    async def message_handler(msg):
        print(f"Received message on {msg.subject}: {msg.data.decode()}")
    
    await nc.subscribe("orders.>", cb=message_handler)
    
    try:
        await asyncio.sleep(600)
    finally:
        await nc.close()

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

Create a publisher script (nats_publisher.py):

#!/usr/bin/env python3
import asyncio
from nats.aio.client import Client as NATS
import json

async def main():
    nc = NATS()
    await nc.connect(servers=["nats://default:securepass123@localhost:4222"])
    
    order = {
        "order_id": "12345",
        "customer": "John Doe",
        "amount": 99.99,
        "status": "created"
    }
    
    await nc.publish("orders.created", json.dumps(order).encode())
    print(f"Published: {json.dumps(order)}")
    
    await nc.close()

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

Run the scripts:

python3 nats_subscriber.py &
python3 nats_publisher.py

Request-Reply Pattern

Request-reply enables synchronous communication. One service sends a request and waits for a reply. Create a request handler service:

#!/usr/bin/env python3
import asyncio
from nats.aio.client import Client as NATS
import json
import uuid

async def main():
    nc = NATS()
    await nc.connect(servers=["nats://default:securepass123@localhost:4222"])
    
    async def request_handler(msg):
        data = json.loads(msg.data.decode())
        print(f"Processing request: {data}")
        
        # Simulate processing
        result = {
            "request_id": data.get("request_id"),
            "status": "completed",
            "result": "Order processed successfully"
        }
        
        await nc.publish(msg.reply, json.dumps(result).encode())
    
    await nc.subscribe("orders.process", cb=request_handler)
    print("Request handler listening on orders.process")
    
    try:
        await asyncio.sleep(600)
    finally:
        await nc.close()

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

Create a request client:

#!/usr/bin/env python3
import asyncio
from nats.aio.client import Client as NATS
import json
import uuid

async def main():
    nc = NATS()
    await nc.connect(servers=["nats://default:securepass123@localhost:4222"])
    
    request_data = {
        "request_id": str(uuid.uuid4()),
        "order_id": "12345",
        "customer": "Jane Smith"
    }
    
    try:
        reply = await nc.request(
            "orders.process",
            json.dumps(request_data).encode(),
            timeout=5
        )
        result = json.loads(reply.data.decode())
        print(f"Reply received: {result}")
    except Exception as e:
        print(f"Request failed: {e}")
    
    await nc.close()

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

JetStream Persistent Storage

JetStream adds persistence and advanced features like message retention and replay. Enable JetStream in configuration (already included in basic config). Create a JetStream stream:

nats stream create ORDERS \
  --subjects "orders.>" \
  --storage file \
  --max-age 30d \
  --server nats://default:securepass123@localhost:4222

List streams:

nats stream list --server nats://default:securepass123@localhost:4222

Create a consumer for the stream:

nats consumer create ORDERS ORDER_PROCESSOR \
  --target order_consumer \
  --deliver latest \
  --ack explicit \
  --server nats://default:securepass123@localhost:4222

Publish to JetStream stream:

nats pub "orders.created" "Order #12345" --server nats://default:securepass123@localhost:4222

Consume from JetStream:

nats consumer next ORDERS ORDER_PROCESSOR --server nats://default:securepass123@localhost:4222

Create a Python client for JetStream:

#!/usr/bin/env python3
import asyncio
from nats.aio.client import Client as NATS
import json

async def main():
    nc = NATS()
    await nc.connect(servers=["nats://default:securepass123@localhost:4222"])
    
    js = nc.jetstream()
    
    # Add a message to the stream
    order = {"order_id": "12345", "status": "pending"}
    await js.publish("orders.created", json.dumps(order).encode())
    
    # Subscribe with consumer
    sub = await js.subscribe("orders.created")
    
    async for msg in sub.messages:
        print(f"Message: {msg.data.decode()}")
        await msg.ack()
    
    await nc.close()

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

NATS Clustering

Create a NATS cluster for high availability and distribution. Configure broker 1 at /etc/nats/broker1.conf:

# Broker 1
port: 4222
http: 8222

jetstream {
  store_dir: /var/lib/nats/broker1
  max_memory: 512M
  max_file: 10GB
}

# Cluster configuration
cluster {
  name: "nats-cluster"
  listen: 0.0.0.0:6222
  routes: [
    nats-route://broker2:6222
    nats-route://broker3:6222
  ]
  auth {
    username: "route_user"
    password: "route_password"
  }
}

authorization {
  users: [
    {user: "default", password: "securepass123"}
  ]
}

log_file: /var/log/nats/broker1.log

Configure broker 2 at /etc/nats/broker2.conf:

# Broker 2
port: 4223
http: 8223

jetstream {
  store_dir: /var/lib/nats/broker2
  max_memory: 512M
  max_file: 10GB
}

cluster {
  name: "nats-cluster"
  listen: 0.0.0.0:6222
  routes: [
    nats-route://broker1:6222
    nats-route://broker3:6222
  ]
  auth {
    username: "route_user"
    password: "route_password"
  }
}

authorization {
  users: [
    {user: "default", password: "securepass123"}
  ]
}

log_file: /var/log/nats/broker2.log

Start all brokers:

nats-server -c /etc/nats/broker1.conf &
nats-server -c /etc/nats/broker2.conf &
nats-server -c /etc/nats/broker3.conf &

Verify cluster formation:

nats server list --server nats://default:securepass123@localhost:4222

Security and Authentication

Implement TLS/SSL for encrypted communication. Generate certificates:

sudo mkdir -p /etc/nats/certs
cd /etc/nats/certs

# Generate self-signed certificates
openssl req -x509 -nodes -days 365 -newkey rsa:2048 \
  -keyout server.key \
  -out server.crt \
  -subj "/CN=nats.example.com"

Update configuration with TLS:

tls: {
  cert_file: /etc/nats/certs/server.crt
  key_file: /etc/nats/certs/server.key
  timeout: 3
}

# Require TLS for client connections
tls_required: true

Create advanced authentication with multiple users:

authorization {
  users: [
    {
      user: "admin"
      password: "admin_secure_password"
      permissions: {
        publish: ">"
        subscribe: ">"
      }
    }
    {
      user: "publisher"
      password: "pub_password"
      permissions: {
        publish: "orders.>"
        subscribe: "_INBOX.>"
      }
    }
    {
      user: "subscriber"
      password: "sub_password"
      permissions: {
        subscribe: "orders.>"
      }
    }
  ]
}

Client Libraries

NATS provides client libraries for multiple languages. Install Python client:

pip3 install asyncio-nats-client

Install Go client:

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

Install Node.js client:

npm install nats

Create a Go client example:

package main

import (
  "fmt"
  "log"
  "github.com/nats-io/nats.go"
)

func main() {
  nc, err := nats.Connect("nats://default:securepass123@localhost:4222")
  if err != nil {
    log.Fatal(err)
  }
  defer nc.Close()

  // Subscribe to subject
  sub, err := nc.Subscribe("orders.>", func(m *nats.Msg) {
    fmt.Printf("Received: %s\n", string(m.Data))
  })
  if err != nil {
    log.Fatal(err)
  }
  defer sub.Unsubscribe()

  // Publish a message
  nc.Publish("orders.created", []byte("Order #12345"))

  select {}
}

Monitoreo

Monitor NATS using the HTTP monitoring port. Access statistics:

curl http://localhost:8222/varz

Get connection details:

curl http://localhost:8222/connz

Monitor server health:

curl http://localhost:8222/healthz

Use nats-top for real-time monitoring:

nats-top -s nats://default:securepass123@localhost:4222

Conclusión

NATS provides a lightweight, high-performance messaging system ideal for cloud-native applications. This guide covered installation, pub-sub and request-reply messaging patterns, JetStream persistence, clustering, security, and monitoring. For production deployments, implement TLS encryption, configure robust authentication, deploy clusters across geographic regions, monitor with dedicated tools, and establish backup procedures for persistent data. NATS excels in scenarios requiring low-latency communication, service discovery, and event-driven architectures at scale.