事件驱动架构的极简落地：用 NATS 替代 Kafka 的决策边界与实践

一、Kafka 的运维噩梦：轻量场景下的架构肥胖

Kafka 是事件驱动架构的事实标准，但它有一个致命问题：太重了。生产环境的 Kafka 集群至少需要 3 个 Broker + 3 个 ZooKeeper 节点。配置、调优、监控、升级——每一项都是专门的运维技能。对于只有几个微服务的团队来说，维护 Kafka 集群本身就是一项全职工作。

NATS 是另一个选择。它是一个用 Go 语言编写的消息系统，部署只需一个不到 20MB 的二进制文件。不需要 ZooKeeper，不需要专门的集群管理工具，不需要 JVM 调优。启动一个三节点的集群只需要：

nats-server --cluster nats://0.0.0.0:6222 --routes nats://node1:6222,nats://node2:6222

但 NATS 不是 Kafka 的平替。它们在设计哲学上有根本差异，知道什么时候该用什么，比会用什么更重要。

graph TB subgraph Kafka[Kafka 适用场景] K1[事件溯源<br/>Event Sourcing] K2[海量数据管道<br/>TB 级日志] K3[消息重放<br/>Replay] K4[多消费者组<br/>独立消费] end subgraph NATS[NATS 适用场景] N1[服务间异步通信] N2[实时推送<br/>毫秒级延迟] N3[请求-回复模式] N4[微服务 < 20] end Q[决策问题: 选哪个?] --> A{消息量级?} A -->|MB/天| NATS A -->|GB- TB/天| Kafka A -->|需要消息持久化<br/>且支持重放?| Kafka A -->|只需要传递消息<br/>不需要历史| NATS A -->|运维资源 < 1 人| NATS style NATS fill:#51cf66,color:#fff style Kafka fill:#4dabf7,color:#fff

本文将分析 NATS 的核心能力、与 Kafka 的差异边界，并通过一个实际的订单事件系统展示极简事件驱动的完整实现。

二、NATS 的核心机制：Subject、JetStream 与 At-Least-Once

NATS 最基础的模式是 Publish-Subscribe。消息通过 Subject（主题）路由，支持通配符：

orders.created → 精确匹配 orders.* → 匹配 orders.created, orders.updated orders.> → 匹配 orders.created, orders.updated.status

NATS 原生提供的是 At-Most-Once 语义（至多一次交付）。如果消费者不在线，消息直接丢失。这对实时推送场景（如 WebSocket 通知）是合适的，但对于订单系统等需要可靠性的场景，这就是问题。

JetStream 是 NATS 的持久化层，提供了 At-Least-Once 语义。它在 NATS 服务器上维护消息持久化存储，消费者可以通过 Stream（类似 Kafka Topic）订阅消息。关键特性：

• 消息保留：基于时间（24 小时）或大小（1GB）的策略
• 消费者组：多个消费者可以共享消费进度
• Ack 机制：消费者确认后才能消费下一条
• 重试策略：未确认的消息自动重发

JetStream 让 NATS 具备了接近 Kafka 的可靠性，但运维复杂度比 Kafka 低一个数量级。

三、NATS + JetStream 的订单事件系统实现

package main import ( "encoding/json" "fmt" "log" "time" "github.com/nats-io/nats.go" ) // OrderEvent 订单事件结构 type OrderEvent struct { OrderID string `json:"order_id"` UserID string `json:"user_id"` Amount float64 `json:"amount"` Status string `json:"status"` Timestamp time.Time `json:"timestamp"` } // Publisher 发布订单事件 type Publisher struct { conn *nats.Conn js nats.JetStreamContext } func NewPublisher(natsURL string) (*Publisher, error) { conn, err := nats.Connect(natsURL, nats.ReconnectWait(2*time.Second), nats.MaxReconnects(10), nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { log.Printf("NATS disconnected: %v", err) }), nats.ReconnectHandler(func(nc *nats.Conn) { log.Println("NATS reconnected") }), ) if err != nil { return nil, fmt.Errorf("connect nats: %w", err) } js, err := conn.JetStream() if err != nil { conn.Close() return nil, fmt.Errorf("create jetstream: %w", err) } return &Publisher{conn: conn, js: js}, nil } func (p *Publisher) PublishOrderCreated(event OrderEvent) error { // 确保 Stream 存在 _, err := p.js.AddStream(&nats.StreamConfig{ Name: "ORDERS", Subjects: []string{"orders.created", "orders.updated", "orders.cancelled"}, MaxAge: 24 * time.Hour, Storage: nats.FileStorage, }, nats.MaxWait(5*time.Second)) if err != nil { log.Printf("stream may already exist: %v", err) } data, err := json.Marshal(event) if err != nil { return fmt.Errorf("marshal event: %w", err) } // 发布到 JetStream, 保证至少一次交付 ack, err := p.js.Publish("orders.created", data) if err != nil { return fmt.Errorf("publish event: %w", err) } log.Printf("Published event %s, seq: %d", event.OrderID, ack.Sequence) return nil } func (p *Publisher) Close() { p.conn.Drain() } // Consumer 消费订单事件 type Consumer struct { conn *nats.Conn js nats.JetStreamContext sub *nats.Subscription } func NewConsumer(natsURL, durableName string) (*Consumer, error) { conn, err := nats.Connect(natsURL, nats.ReconnectWait(2*time.Second), nats.MaxReconnects(-1), ) if err != nil { return nil, fmt.Errorf("connect nats: %w", err) } js, err := conn.JetStream() if err != nil { conn.Close() return nil, fmt.Errorf("create jetstream: %w", err) } // 创建持久化消费者组 sub, err := js.PullSubscribe( "orders.created", durableName, // 持久化名称，重启后保留消费进度 nats.AckExplicit(), nats.MaxDeliver(3), // 最多重试 3 次 nats.AckWait(30*time.Second), // 30 秒内必须 ACK ) if err != nil { conn.Close() return nil, fmt.Errorf("subscribe: %w", err) } return &Consumer{conn: conn, js: js, sub: sub}, nil } func (c *Consumer) ProcessMessages(handler func(OrderEvent) error) error { for { // 批量拉取消息 msgs, err := c.sub.Fetch(10, nats.MaxWait(5*time.Second)) if err != nil { if err == nats.ErrTimeout { continue // 无消息, 等待 } return fmt.Errorf("fetch messages: %w", err) } for _, msg := range msgs { var event OrderEvent if err := json.Unmarshal(msg.Data, &event); err != nil { log.Printf("unmarshal event: %v", err) msg.Nak() // 消息格式错误，不重试 continue } if err := handler(event); err != nil { log.Printf("handle event %s: %v", event.OrderID, err) msg.Nak() // 处理失败，标记重试 continue } msg.Ack() // 处理成功，确认 } } } func (c *Consumer) Close() { if c.sub != nil { c.sub.Unsubscribe() } c.conn.Drain() } // 使用示例 func main() { publisher, err := NewPublisher("nats://localhost:4222") if err != nil { log.Fatal(err) } defer publisher.Close() // 发布订单事件 event := OrderEvent{ OrderID: "ORD-2024-001", UserID: "USR-123", Amount: 99.99, Status: "created", Timestamp: time.Now(), } if err := publisher.PublishOrderCreated(event); err != nil { log.Fatal(err) } }

消费者使用示例：

consumer, err := NewConsumer("nats://localhost:4222", "order-service") if err != nil { log.Fatal(err) } defer consumer.Close() err = consumer.ProcessMessages(func(event OrderEvent) error { log.Printf("Processing order: %s, amount: %.2f", event.OrderID, event.Amount) // 实际的业务逻辑: 更新库存、发送通知等 if event.Amount <= 0 { return fmt.Errorf("invalid amount") } return nil })

四、NATS 与 Kafka 的决策矩阵

选择 NATS 的标准：

• 服务数量 < 20
• 不需要消息重放历史
• 开发/运维资源有限
• 对延迟敏感（如实时通知）

选择 Kafka 的标准：

• 消息数据需要长期保留（> 7 天）
• 需要独立的多消费者组按不同速度消费
• 消息量达到 TB 级别/天
• 团队已有 Kafka 运维经验

五、总结

NATS 是事件驱动架构中的"极简选项"。它用最小的运维成本实现了微服务间的可靠异步通信。但极简不是万能的——如果你的业务需要事件溯源、消息重放或海量数据管道，Kafka 才是正确的选择。

落地路径：先用 NATS 的原生 Pub/Sub 做服务间异步通信；确认需要消息持久化后，升级到 JetStream；如果日后消息量突破 GB/天级别，再评估是否迁移到 Kafka。

技术的选择不是看哪个"更好"，而是看哪个"更适合当前的约束条件"。少即是多——用 NATS 而不是 Kafka，不是因为 NATS 更强大，而是它刚好够用。