基于NATS实现多租户命令消息按租户串行处理的最优方案

需求回顾 #

• 多租户系统中，同一租户的命令需串行处理，保证命令与对应事件序列有序，可忽略OCC、版本控制等机制
• 支持部署多命令处理器实例实现扩容与高可用，同一时刻每个租户仅能有一个命令在处理
• 租户动态创建/删除，需避免为每个租户单独配置消费者的繁琐操作

原方案问题分析 #

• 设置MaxAckPending(1)会作用于整个队列消费者组，导致所有租户同一时刻仅能处理一个消息，不同租户无法并行处理，完全不符合扩容需求
• 为每个租户单独订阅tenant-{id}.commands的方式，无法适配动态租户的创建与删除，运维成本极高

方案一：消息分组（GroupBy）+ 按分组限制未确认消息数（推荐） #

利用JetStream的GroupBy特性，按租户ID对消息分组，同时为每个分组设置MaxAckPendingPerGroup(1)，确保同一租户的消息串行处理，不同租户的消息可并行处理，完美适配动态租户场景。

实现代码示例 #

import "strings"

// 从消息主题中提取租户ID作为分组键（假设主题格式为 {tenantId}.commands）
groupByTenant := func(msg *nats.Msg) string {
    parts := strings.Split(msg.Subject, ".")
    if len(parts) > 0 {
        return parts[0]
    }
    // 处理异常主题，可分配到默认分组或直接丢弃
    return "default"
}

// 创建带分组配置的队列消费者
consumer, err := js.QueueSubscribe(
    "*.commands",          // 订阅所有租户的命令主题
    "cmd-handler-group",   // 队列组名称，多实例加入同一组实现负载均衡
    func(msg *nats.Msg) {
        // 执行命令处理逻辑
        // ...
        
        // 处理完成后确认消息，触发下一条同租户消息的分发
        msg.Ack()
    },
    nats.GroupBy(groupByTenant),               // 按租户ID分组
    nats.MaxAckPendingPerGroup(1),             // 每个租户分组最多1个未确认消息，保证串行
    nats.Durable("cmd-handler-durable"),       // 持久化消费者，重启后恢复状态
    nats.AckWait(time.Second*30),              // 设置消息确认超时，避免死锁
)

方案优势 #

• 完全适配动态租户：无需提前配置，新租户的命令自动纳入分组逻辑
• 资源高效利用：不同租户的命令可在多实例间并行处理，仅同一租户串行
• 可靠性保障：持久化消费者确保重启后不丢失未处理消息

方案二：分区流（Stream Partitioning）（适合超大规模租户场景） #

如果租户数量达到数万级以上，可采用分区流方案，按租户ID哈希值将消息分配到固定数量的分区中，每个分区内的消息天然有序，再通过队列消费实现多实例扩容。

实现步骤 #

1. 创建分区流：

streamCfg := &nats.StreamConfig{
    Name:           "commands-stream",
    Subjects:       []string{"*.commands"},
    Storage:        nats.FileStorage,
    NumPartitions:  20,  // 根据租户规模设置分区数，建议为处理器实例数的2-4倍
    Partitioning:   nats.PartitionBySubject,  // 按主题哈希分区，同一租户的消息会进入同一分区
}

_, err := js.AddStream(streamCfg)

2. 创建队列消费者：

_, err := js.QueueSubscribe(
    "*.commands",
    "cmd-handler-group",
    func(msg *nats.Msg) {
        // 命令处理逻辑
        msg.Ack()
    },
    nats.MaxAckPending(1),  // 每个分区的消费者最多1个未确认消息，保证分区内串行
    nats.Durable("cmd-handler-durable"),
)

方案优势与局限 #

• 优势：分区级隔离，性能更高，适合超大规模租户场景
• 局限：分区数量创建时固定，无法动态调整，需提前规划容量

关键结论 #

• 优先选择方案一，适配动态租户场景，配置简单且资源利用率高
• 原方案中MaxAckPending(1)是全局限制，会导致所有租户串行，不可采用
• 无需为每个租户单独创建消费者，通过分组或分区即可实现租户级串行处理

内容的提问来源于stack exchange，提问作者Twice_Twice