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

事件驱动架构的极简落地：用 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 更强大，而是它刚好够用。