astronomerinc/ap-nats-streaming

Astronomer平台NATS Streaming镜像文档

镜像概述

本镜像为Astronomer平台定制，基于NATS Streaming（STAN）构建。NATS Streaming是NATS消息系统的扩展，提供持久化、消息重播、事件流处理等增强功能，专为需要可靠消息传递的分布式系统设计。该镜像预配置适配Astronomer平台架构，可无缝集成到平台生态中，作为核心消息传递层支持组件间通信和事件驱动工作流。

核心功能与特性

1. 高性能消息传递

• 支持多种消息模式：发布/订阅（Pub/Sub）、请求/回复（Request/Reply）、队列订阅（Queue Subscriptions）
• 低延迟消息处理，吞吐量可达每秒数十万消息
• 轻量级设计，资源占用低，启动快速

2. 消息持久化与可靠性

• 支持消息持久化存储（基于文件系统或内存）
• 消息确认机制（ACK）确保消息可靠传递
• 自动重连功能，保障连接稳定性

3. 消息流高级特性

• 消息重播：支持按时间范围、序列号或最后位置重播消息
• 持久化订阅：即使订阅者离线，消息也会保存直至被处理
• 消息限流：防止消费者被消息洪流淹没

4. 可扩展性与集成

• 支持集群模式部署，实现水平扩展
• 与Astronomer平台组件（如Airflow、Houston API）预集成
• 兼容标准NATS客户端协议，支持多语言接入

使用场景

• Astronomer平台内部通信：连接平台各微服务（如元数据管理、任务调度、用户认证）
• 事件驱动工作流：触发基于事件的自动化流程（如DAG部署、任务状态变更通知）
• 数据流处理：收集、传输和处理平台运行时数据（如任务执行日志、资源使用指标）
• 微服务解耦：作为消息中间件，降低服务间直接依赖，提高系统弹性
• 灾备与数据恢复：利用消息持久化和重播功能，支持系统故障后的状态恢复

使用方法与配置

基本运行

使用以下命令启动基本的NATS Streaming容器：

bash
docker run -d \
  --name astronomer-nats-streaming \
  -p 4222:4222 \  # NATS客户端连接端口
  -p 8222:8222 \  # 监控端口
  astronomerinc/nats-streaming:latest

自定义配置

通过环境变量或配置文件进行自定义配置，常用环境变量如下：

示例：自定义存储路径和集群ID

bash
docker run -d \
  --name astronomer-nats-streaming \
  -p 4222:4222 \
  -p 8222:8222 \
  -v /host/path/to/store:/data/stan/store \  # 挂载宿主机目录作为持久化存储
  -e STORE_DIR=/data/stan/store \
  -e CLUSTER_ID=astronomer-production \
  -e MAX_PAYLOAD_SIZE=2097152 \  # 2MB
  astronomerinc/nats-streaming:latest

Docker Compose配置

在Astronomer平台部署中，建议使用Docker Compose管理NATS Streaming服务：

yaml
version: '3.8'

services:
  nats-streaming:
    image: astronomerinc/nats-streaming:latest
    container_name: astronomer-nats-streaming
    restart: always
    ports:
      - "4222:4222"
      - "8222:8222"
    environment:
      - NATS_LISTEN_ADDRESS=0.0.0.0
      - STORE_TYPE=file
      - STORE_DIR=/data/stan/store
      - CLUSTER_ID=astronomer-stan-cluster
      - RETENTION_POLICY=limits
      - MAX_AGE=72h  # 消息最大保留时间（3天）
    volumes:
      - nats-store:/data/stan/store
    networks:
      - astronomer-platform-network

volumes:
  nats-store:  # 持久化卷，确保数据不丢失

networks:
  astronomer-platform-network:
    external: true  # 假设已存在平台网络

集群部署

对于高可用需求，可部署NATS Streaming集群：

bash
# 节点1
docker run -d \
  --name nats-streaming-node1 \
  -e CLUSTER_ID=astronomer-cluster \
  -e NATS_LISTEN_ADDRESS=0.0.0.0 \
  -e ROUTES=nats-route://nats-streaming-node2:6222,nats-route://nats-streaming-node3:6222 \
  -p 4222:4222 \
  -p 6222:6222 \  # 集群路由端口
  astronomerinc/nats-streaming:latest

# 节点2（类似配置，调整ROUTES指向其他节点）
# 节点3（类似配置，调整ROUTES指向其他节点）

注意事项

• 持久化存储需确保磁盘空间充足，避免消息存储满导致服务异常
• 集群部署时，所有节点的CLUSTER_ID必须一致
• 生产环境建议启用认证机制（如TLS加密、令牌认证），增强安全性
• 监控端口（8222）默认开放，生产环境可限制访问来源或禁用
• 定期备份持久化数据，防止数据丢失

客户端连接

NATS Streaming兼容标准NATS客户端，示例Python客户端连接代码：

python
import asyncio
from nats.aio.client import Client as NATS
from stan.aio.client import Client as STAN

async def run():
    # 连接NATS服务器
    nc = NATS()
    await nc.connect("nats://nats-streaming:4222")
    
    # 连接NATS Streaming
    sc = STAN()
    await sc.connect("astronomer-stan-cluster", "client-123", nats=nc)
    
    # 订阅消息
    async def callback(msg):
        print(f"Received message: {msg.data.decode()}")
    
    await sc.subscribe("platform.events", queue="worker-queue", cb=callback)
    
    # 发布消息
    await sc.publish("platform.events", b"Task completed successfully")

if __name__ == '__main__':
    asyncio.run(run())