NATS Core Fundamentals

Session 1 - Messaging Patterns

pub/sub request/reply queue groups subjects

What is NATS?

A connective technology for modern distributed systems. Simple, fast, and lightweight messaging.

Why NATS?

Subject-Based Addressing

Messages are published to subjects, not queues or topics.

Subject Hierarchy

Use . as delimiter to create hierarchies:

company.department.action

orders.new
orders.fulfilled
orders.cancelled

payments.processed
payments.failed

Enables powerful wildcard subscriptions

Wildcards

orders.*
  orders.new       OK
  orders.fulfilled OK
  orders.us.new    NO

orders.>
  orders.new         OK
  orders.fulfilled   OK
  orders.us.east.new OK

> must be the last token

Pattern 1: Publish/Subscribe

All subscribers receive every message (fan-out)

Pub/Sub in Python

import asyncio
import nats

async def main():
    nc = await nats.connect("nats://localhost:4222")
    
    async def handler(msg):
        print(f"Received: {msg.data.decode()}")
    
    await nc.subscribe("orders.>", cb=handler)
    await nc.publish("orders.new", b"Order #123")
    
    await asyncio.sleep(1)
    await nc.close()

asyncio.run(main())

Pattern 2: Queue Groups

Only ONE subscriber receives each message (load balancing)

Queue Groups in Python

async def worker(nc, worker_id):
    async def handler(msg):
        print(f"Worker {worker_id}: {msg.data.decode()}")
    
    # Join queue group "workers"
    await nc.subscribe("tasks", queue="workers", cb=handler)

# NATS automatically load-balances between workers

Pattern 3: Request/Reply

Synchronous RPC over async messaging

Request/Reply in Python

# Service
async def service(nc):
    async def handler(msg):
        response = f"Processed: {msg.data.decode()}"
        await nc.publish(msg.reply, response.encode())
    
    await nc.subscribe("api.process", cb=handler)

# Client
resp = await nc.request("api.process", b"data", timeout=1.0)
print(f"Got: {resp.data.decode()}")

Message Structure

Subject:    orders.new              (required)
Reply-To:   _INBOX.xyz123           (optional)
Headers:    {"X-Trace-Id": "abc"}   (optional)
Payload:    {"id": 123, ...}        (bytes, max 1MB)

• Subject: Address for routing
• Reply-To: For request/reply pattern
• Headers: Metadata (compression, tracing)
• Payload: Your data (JSON, Protobuf, etc.)

Core NATS: At-Most-Once

• Messages delivered to active subscribers only
• No persistence - if no one listening, message is gone
• No acknowledgments in core NATS
• Extremely fast (14M+ msg/sec)

Need persistence? That is JetStream (Session 2)

Connection Management

nc = await nats.connect(
    servers=["nats://localhost:4222"],
    reconnect_time_wait=2,
    max_reconnect_attempts=-1,
    ping_interval=20,
)

nc.on_disconnect = lambda: print("Disconnected!")
nc.on_reconnect = lambda: print("Reconnected!")

await nc.drain()  # Graceful shutdown

Error Handling

from nats.errors import TimeoutError, NoRespondersError

try:
    resp = await nc.request("api.users", b"get", timeout=1.0)
except TimeoutError:
    print("Service did not respond in time")
except NoRespondersError:
    print("No service listening on this subject")

Three Core Patterns

Mix and match: queue group of services handling requests

Real-World: Payment Gateway

Subjects:
  payments.initiated
  payments.{provider}.result
  payments.completed
  payments.failed

Queue Groups:
  payments.initiated  queue="processors"

Request/Reply:
  api.balance.check   Balance Service

Interactive Demo

Try pub/sub, wildcards, queue groups, and request/reply live:

Connected to live NATS server at wss://i33ym.cc/ws/nats

Summary

Key Takeaway

Resources

• Essay: NATS Core Deep Dive
• Interactive Demo
• Official NATS Documentation
• nats.py - Python Client

Next Session

JetStream: Persistence and Streaming

• Streams and message persistence
• Consumers: push vs pull
• At-least-once and exactly-once delivery
• Replay and acknowledgments

Questions?

NATS Core Fundamentals - Session 1