事件驱动架构的极简落地用 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-OnceNATS 最基础的模式是 Publish-Subscribe。消息通过 Subject主题路由支持通配符orders.created → 精确匹配 orders.* → 匹配 orders.created, orders.updated orders. → 匹配 orders.created, orders.updated.statusNATS 原生提供的是 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 的决策矩阵维度NATSKafka部署复杂度单二进制秒级启动需要 ZooKeeper/KRaft Broker分钟级运维成本几乎为零需要专人消息持久化JetStream 支持原生支持消息重放有限Stream 内重放完整支持按 Offset 重放延迟 1ms~10ms吞吐量数百万 msg/s数百万 msg/s事件溯源勉强可用原生设计资源占用 50MB 内存 1GB 内存JVM选择 NATS 的标准服务数量 20不需要消息重放历史开发/运维资源有限对延迟敏感如实时通知选择 Kafka 的标准消息数据需要长期保留 7 天需要独立的多消费者组按不同速度消费消息量达到 TB 级别/天团队已有 Kafka 运维经验五、总结NATS 是事件驱动架构中的极简选项。它用最小的运维成本实现了微服务间的可靠异步通信。但极简不是万能的——如果你的业务需要事件溯源、消息重放或海量数据管道Kafka 才是正确的选择。落地路径先用 NATS 的原生 Pub/Sub 做服务间异步通信确认需要消息持久化后升级到 JetStream如果日后消息量突破 GB/天级别再评估是否迁移到 Kafka。技术的选择不是看哪个更好而是看哪个更适合当前的约束条件。少即是多——用 NATS 而不是 Kafka不是因为 NATS 更强大而是它刚好够用。