你的具体版本，上下文最大窗口，知识库最新更新日期

Question

Answer 1

我的具体信息如下：

项目	答案
具体模型版本	GPT-5.5 Thinking
上下文最大窗口	我这里没有被提供一个可对外确认的精确 token 数，所以不能负责任地报具体数值。实际可用上下文也会受当前产品环境、工具、附件和会话状态影响。
知识库最新更新日期	2025 年 8 月
当前系统日期	2026 年 6 月 17 日
你的时区	Europe/Tallinn

所以：我能确定的是模型名和知识截止时间；最大上下文窗口的精确数值我不能臆测。

Answer 2

下面按 3 问分别回答。核心结论是：不要再把 10,000 人大群消息同步 fanout 到 10,000 个用户 ZSET；应改成“大群消息一份持久日志 + 用户游标”的 fanout-on-read 模型。

① 根因分析：Cluster 下 pipeline 为什么不是“真正一次批量执行”

Redis pipeline 的本质是：在一个 TCP 连接上连续发送多条命令，不等每条命令的响应，最后再批量读取响应。它减少的是“每条命令一次 RTT”的等待成本，不是把多条命令变成 Redis 服务端的一条原子批处理命令。Redis 官方文档也明确说明，pipeline 是让客户端连续发送多条命令，最后一次性读取响应。(Redis)

但 Redis Cluster 不是一个会帮你转发命令的中心代理。Redis Cluster 中每个 master 负责 16384 个 hash slot 的一个子集；客户端通常维护 slot 到节点的映射，并直接把命令发给负责该 slot 的 master。Redis 官方 Cluster 规范说明，正常情况下客户端会直接联系正确节点，而不是让 Redis 节点代理转发。(Redis)

所以你的 10,000 条：

text
ZADD offline:{uid} score msg

实际不会形成：

text
client -> cluster: 一次性 10,000 条 ZADD
cluster -> 自动分发到 3 个 master

而是形成：

text
client
  -> master-1 TCP connection: 约 3333 条 ZADD
  -> master-2 TCP connection: 约 3333 条 ZADD
  -> master-3 TCP connection: 约 3334 条 ZADD

更精确的数据流是：

text
for uid in groupMembers:
    key = "offline:" + uid
    slot = CRC16(key) % 16384
    node = slotOwner[slot]

    serialize RESP:
        *4
        ZADD
        key
        score
        msgId/msgPayload

    append/write to node 对应的 TCP connection
    在本地 responseQueue[node] 里登记一个 Response

然后 sync() 阶段：

text
for each node connection:
    read N_node 个 ZADD 响应
    反序列化
    填充到对应 Response

Jedis 的 ClusterPipeline 说明里也强调它是 Cluster 环境下把命令自动路由到对应节点，并且只支持可路由到单节点的命令；跨节点命令不是一个真正的单节点 pipeline。(GitHub) 旧版/部分 Jedis 多节点 pipeline 实现中，源码层面也能看到它按 nodeKey 取连接、按连接维护 response queue，并在 sync() 时对各节点连接读取响应。(Google Cloud Storage)

因此，这 10,000 条 ZADD 在网络层实际是 3 条 TCP 连接上的 3 组 pipeline，不是一次发给 Redis Cluster 的服务端批处理。Redis 每个 master 收到自己的那一组命令后，仍然在该 master 的事件循环中逐条执行 ZADD。pipeline 省掉的是“逐条等待响应”，没有省掉“10,000 次命令解析、10,000 次 ZSET 修改、10,000 次响应编码”。

如果客户端 slot 缓存过期，还会发生：

text
client -> wrong master: ZADD key
wrong master -> client: MOVED slot correctHost:port
client 刷新 slot 映射
client -> correct master: 重发 ZADD

Redis Cluster 对 MOVED/ASK 重定向的行为也是官方规范的一部分。(Redis)

② 量化推导：理论最小耗时与 480ms 意味着什么

假设：

text
群成员数 N = 10,000
master 数 M = 3
key 均匀分布
RTT = 0.3ms

那么每个 master 大约收到：

text
N / M ≈ 3333 条 ZADD

情况 A：智能 Cluster pipeline 按 master 聚合

正常 smart client 最终应按目标 master 聚合，而不是每个 uid 一次 RTT。

如果 3 个 master 的 pipeline 串行发送/读取：

text
T_network_min = 3 * RTT = 3 * 0.3ms = 0.9ms

如果 3 个 master 的 pipeline 并行发送/读取：

text
T_network_min = 1 * RTT = 0.3ms

所以，仅从 RTT 角度看，Cluster pipeline 的理论网络下界是：

text
0.3ms ~ 0.9ms

这还没算 Redis 执行 10,000 次 ZADD 的 CPU 和内存分配成本，但它说明一件事：480ms 绝不主要来自 0.3ms RTT。

当前观测：

text
T_observed_pipeline = 480ms
T_network_min_serial_by_master = 0.9ms
T_extra ≈ 480ms - 0.9ms = 479.1ms
放大倍数 ≈ 480 / 0.9 ≈ 533 倍

换算成吞吐：

text
10,000 条 ZADD / 0.48s ≈ 20,833 commands/s
每个 master ≈ 3333 / 0.48s ≈ 6,944 commands/s

对于 pipeline 场景，这个吞吐明显偏低。Redis 官方也说明 pipeline 不只是减少 RTT，还能减少大量 socket I/O syscall 成本；因此正常 pipeline 吞吐不应该被 0.3ms RTT 限制在这个级别。(Redis)

情况 B：如果真的“按 slot 串行发送”

Redis Cluster 有 16384 个 slot。10,000 个随机 uid 落入 slot 的期望占用数是：

text
E(occupiedSlots)
= 16384 * (1 - (1 - 1/16384)^10000)
≈ 7480 个 slot

如果客户端真的每个 slot 一组，并且每个 slot 串行一个 RTT：

text
T_min_by_slot = 7480 * 0.3ms ≈ 2244ms

但你测得是 480ms，小于 2.2s，所以实际行为大概率不是“每 slot 一次网络 RTT 串行”，而是至少合并到了 master 连接级别，或者命令在构造阶段就已经写入不同节点连接的缓冲区。

480ms 的真实瓶颈判断

结合你给的线程配置：

text
Jetty 线程池 = 990
pipelineExecutor corePoolSize = 4
单条大群消息 pipeline 耗时 = 480ms

这个模型的系统吞吐上限大约是：

text
4 / 0.48s ≈ 8.33 条大群消息/s

也就是系统最多只能同时处理 4 个“大群 fanout Redis 写入任务”。一旦大群消息进入速度接近或超过 8 条/s，pipelineExecutor 队列就开始堆积，Jetty 线程会阻塞等待结果，990 个 Jetty 线程反而会放大排队、上下文切换、对象分配和 GC 压力。

所以 480ms 主要说明：

text
不是 RTT 瓶颈；
是 O(群人数) 写放大 + 客户端 pipeline 构造/序列化/响应解析 + Redis ZSET 写入 CPU/内存成本 + pipelineExecutor 并发度过低共同造成。

特别是当前架构中，一条群消息 = 10,000 次 Redis 写。即使 pipeline 做得再好，它也只是把 10,000 次写“打包传输”，没有消除 10,000 次写本身。

③ 架构重设计：把 10,000 人大群 Redis 写入降到 50ms 内

设计目标

把当前模型：

text
一条群消息 -> 写 10,000 个 user offline ZSET

改成：

text
一条群消息 -> 写 1 条 group message log
每个用户离线消息 -> 通过 group cursor 拉取

也就是：

text
fanout-on-write  ->  fanout-on-read

大群消息只写一份：

text
im:g:{gid}:stream      // 群消息日志，Redis Stream 或 ZSET
im:g:{gid}:seq         // 群内递增 seq
im:g:{gid}:dedup       // msgId 去重索引
im:u:{uid}:gcur        // 用户在每个群的已读/已拉取游标
im:g:{gid}:ack         // 成员 cursor 聚合，用于 GC

注意 key 里的 {gid} 是 Redis hash tag，保证同一个群的 seq、stream、dedup key 落到同一个 hash slot，从而同一个 Lua 脚本可以在 Redis Cluster 中原子执行。Redis Cluster 支持同 slot 多 key 操作；跨 slot 多 key 则不能作为单节点命令执行。(Redis)

写入路径：O(1) Redis 写

发送一条大群消息时：

text
1. 校验发送者是否在群内
2. 生成 msgId
3. Lua 原子执行：
   - 检查 msgId 是否已存在，防止重试重复写
   - INCR 群 seq
   - XADD 一条消息到群 Stream
   - HSET msgId -> streamId/seq
4. 同一条 Redis 连接上执行 WAITAOF 1 1 timeout
5. 成功后 ack 发送方
6. 异步通知在线网关；通知失败也不丢，因为消息已经在群日志里

Redis Streams 原生支持 XADD 写入消息日志，也支持按 ID 范围读取历史消息。(Redis) 为了“消息不丢”，不能只依赖 Redis Cluster 默认异步复制，因为 Redis Cluster 官方规范明确提到，master 写入尚未复制到 replica 时发生故障，已确认写入仍可能丢失。(Redis)

更强的做法是使用 Redis 7.2+ 的 WAITAOF：它会阻塞当前客户端，直到当前连接之前的写命令被本地 AOF fsync，并被指定数量 replica 的 AOF fsync 确认；客户端必须检查返回值是否满足要求。(Redis) 如果没有 Redis 7.2，只能退化为 WAIT 1 timeout，但 WAIT 只能保证副本收到写入，官方也说明它不能让 Redis 变成强一致存储，只是提升实际安全性。(Redis)

Java 关键代码骨架

下面是骨架代码，重点是：Lua 写入和 WAITAOF 必须在同一条 Redis 连接上执行，因为 WAITAOF 等待的是“当前连接之前的写命令”。

java
public final class LargeGroupMessageWriter {

    private static final String APPEND_LUA = """
        -- KEYS[1] = im:g:{gid}:seq
        -- KEYS[2] = im:g:{gid}:stream
        -- KEYS[3] = im:g:{gid}:dedup
        -- ARGV[1] = msgId
        -- ARGV[2] = fromUid
        -- ARGV[3] = body
        -- ARGV[4] = sendTimeMillis

        local existed = redis.call('HGET', KEYS[3], ARGV[1])
        if existed then
            return {'DUP', existed}
        end

        local seq = redis.call('INCR', KEYS[1])

        local streamId = redis.call(
            'XADD', KEYS[2], '*',
            'seq', seq,
            'msgId', ARGV[1],
            'fromUid', ARGV[2],
            'body', ARGV[3],
            'ts', ARGV[4]
        )

        local value = streamId .. ':' .. seq
        redis.call('HSET', KEYS[3], ARGV[1], value)

        return {'OK', value}
        """;

    private final RedisSlotRouter slotRouter;
    private final String appendSha;

    public LargeGroupMessageWriter(RedisSlotRouter slotRouter, String appendSha) {
        this.slotRouter = slotRouter;
        this.appendSha = appendSha;
    }

    public AppendResult append(String gid, String fromUid, String body) {
        String tag = "{g:" + gid + "}";

        String seqKey = "im:g:" + tag + ":seq";
        String streamKey = "im:g:" + tag + ":stream";
        String dedupKey = "im:g:" + tag + ":dedup";

        String msgId = generateMsgId(gid, fromUid);
        String now = Long.toString(System.currentTimeMillis());

        // 必须拿到 streamKey 所属 master 的“同一条物理连接”
        try (RedisConnection conn = slotRouter.masterConnectionForKey(streamKey)) {

            Object luaRet = conn.evalsha(
                appendSha,
                List.of(seqKey, streamKey, dedupKey),
                List.of(msgId, fromUid, body, now)
            );

            // fence 写入用于重试场景：
            // 如果上次写成功但 WAITAOF 超时，下一次重试命中 dedup 后，
            // 仍然写一个 fence，再 WAITAOF，确保复制/AOF offset 追上。
            String fenceKey = "im:g:" + tag + ":fence:" + msgId + ":" + now;
            conn.set(fenceKey, "1", SetOption.px(86_400_000));

            // Redis 7.2+:
            // WAITAOF 1 1 40
            // 含义：等待本地 AOF fsync 1 份 + replica AOF fsync 1 份，最多 40ms。
            WaitAofResult durable = conn.waitAof(1, 1, 40);

            if (durable.localFsyncs() < 1 || durable.replicaFsyncs() < 1) {
                // 不 ack 发送方，让上层用相同 msgId 重试。
                // 因为 Lua 有 dedup，重试不会产生重复消息。
                throw new NotDurableException("message not durable enough: " + durable);
            }

            return AppendResult.fromLua(msgId, luaRet);
        }
    }

    private String generateMsgId(String gid, String fromUid) {
        return gid + ":" + fromUid + ":" + java.util.UUID.randomUUID();
    }
}

这里的 RedisSlotRouter 是你需要封装的 Cluster 路由器：

java
public interface RedisSlotRouter {

    /**
     * 返回 key 所属 Redis Cluster master 的独占连接。
     *
     * 注意：
     * 1. evalsha、fence set、waitaof 必须在同一条连接上。
     * 2. 不要用每条命令都可能换连接的高级封装来做 WAITAOF。
     */
    RedisConnection masterConnectionForKey(String key);
}

RedisConnection 可以基于 Jedis 的底层连接封装，也可以用 Lettuce async connection 封装。关键不是客户端名字，而是连接语义必须正确。

拉取路径：用户游标，不再写 10,000 个 ZSET

用户上线或拉离线消息时：

java
public final class LargeGroupMessagePuller {

    private final RedisSlotRouter slotRouter;

    public List<GroupMessage> pull(String uid, String gid, int limit) {
        String tag = "{g:" + gid + "}";

        String streamKey = "im:g:" + tag + ":stream";
        String userCursorKey = "im:u:" + uid + ":gcur";
        String groupAckKey = "im:g:" + tag + ":ack";

        String lastStreamId = getUserCursor(uid, gid);
        if (lastStreamId == null) {
            lastStreamId = "0-0";
        }

        try (RedisConnection conn = slotRouter.masterConnectionForKey(streamKey)) {
            List<StreamRecord> records = conn.xrangeAfter(streamKey, lastStreamId, limit);

            List<GroupMessage> messages = records.stream()
                .map(GroupMessage::fromStreamRecord)
                .toList();

            return messages;
        }
    }

    public void ackDelivered(String uid, String gid, String streamId, long seq) {
        String tag = "{g:" + gid + "}";

        String userCursorKey = "im:u:" + uid + ":gcur";
        String groupAckKey = "im:g:" + tag + ":ack";

        // cursor 更新发生在用户真实收到/客户端 ack 后。
        // 如果客户端断线，cursor 不推进，下次会重拉，靠 msgId 去重。
        try (RedisConnection conn = slotRouter.masterConnectionForKey(groupAckKey)) {
            conn.hset(userCursorKey, gid, streamId);
            conn.zadd(groupAckKey, seq, uid);
        }
    }

    private String getUserCursor(String uid, String gid) {
        // 可用本地缓存 + Redis HGET
        return null;
    }
}

这会把发送路径的写放大从：

text
10,000 次 ZADD

变成：

text
1 次 Lua append + 1 次 fence SET + 1 次 WAITAOF

也就是 O(1)。

GC：不能按时间随便删，否则会丢离线消息

如果要严格保证大群离线消息不丢，不能简单 XTRIM MAXLEN 10000。正确做法是按所有成员的最小 cursor 清理：

text
im:g:{gid}:ack
score = 用户已确认 seq
member = uid

定时任务：

java
public final class GroupLogGcJob {

    private final RedisSlotRouter slotRouter;

    public void gc(String gid) {
        String tag = "{g:" + gid + "}";

        String streamKey = "im:g:" + tag + ":stream";
        String groupAckKey = "im:g:" + tag + ":ack";

        try (RedisConnection conn = slotRouter.masterConnectionForKey(streamKey)) {
            Optional<CursorScore> minAck = conn.zrangeMinWithScore(groupAckKey);

            if (minAck.isEmpty()) {
                return;
            }

            long minSeq = minAck.get().score();

            // 实际实现中需要 seq -> streamId 的映射。
            // 可以在 stream field 中保存 seq，并维护 seqId index。
            String trimBeforeStreamId = conn.lookupStreamIdBySeq(gid, minSeq);

            conn.xtrimMinId(streamKey, trimBeforeStreamId);
        }
    }
}

如果产品允许“超过 30 天未上线不保证完整历史”，可以按产品策略 trim；但那就不是严格“不丢”。严格不丢意味着：只删所有成员都已经确认过的消息。

新线程模型

建议改成下面这样：

text
Jetty IO / request threads
    只做鉴权、参数校验、快速提交
    不做 10,000 人 Redis fanout

LargeGroupWriteExecutor
    按 gid 或 Redis master 分片
    每个分片少量固定线程 / async event loop
    每条消息只写 1 份 group log

OnlinePushExecutor
    只负责在线用户实时推送
    推送失败不影响可靠性，因为客户端可从 group log 拉取

OfflinePullExecutor
    用户上线或主动 pull 时读取 group stream
    更新用户 cursor

GcExecutor
    后台根据 group 最小 cursor 做 XTRIM

关键是：发送路径不再依赖 10,000 人 fanout 完成。在线推送只是加速实时体验，可靠性来自 Redis 中的 group log + cursor。

为什么可以进 50ms

改造前：

text
1 条大群消息 = 10,000 次 ZADD
耗时 ≈ 480ms

改造后：

text
1 条大群消息 =
    Lua: INCR + XADD + HSET
    SET fence
    WAITAOF 1 1 40

网络上最多是 1 到 2 次 Redis 往返加一个持久化等待。按你给的 RTT=0.3ms，纯网络部分远低于 5ms；主要耗时变成 AOF fsync 和 replica fsync。在同机房/同 AZ、Redis 和磁盘正常的情况下，50ms 是合理目标。

但要明确一点：低延迟和强持久化不能无条件同时保证。如果 WAITAOF 1 1 40 超时，你不应该 ack 发送方，而是返回“发送中/稍后重试”，并用相同 msgId 重试。这样牺牲的是这一次请求的成功率，而不是消息可靠性。

最终方案一句话概括：

text
大群不要写 10,000 个用户离线 ZSET；
改为写 1 条 Redis group stream，
用户用 cursor 拉取，
用 WAITAOF/WAIT 做可靠性边界，
用异步在线推送做实时性优化。

这样才能从根上把 480ms 的 O(N) 写放大，变成 50ms 内可控的 O(1) 写入路径。

你的具体版本，上下文最大窗口，知识库最新更新日期

Question

Réflexion

Réponse

Question

Réflexion

Réflexion

Réponse

① 根因分析：Cluster 下 pipeline 为什么不是“真正一次批量执行”

② 量化推导：理论最小耗时与 480ms 意味着什么

情况 A：智能 Cluster pipeline 按 master 聚合

情况 B：如果真的“按 slot 串行发送”

480ms 的真实瓶颈判断

③ 架构重设计：把 10,000 人大群 Redis 写入降到 50ms 内

设计目标

写入路径：O(1) Redis 写

Java 关键代码骨架

拉取路径：用户游标，不再写 10,000 个 ZSET

GC：不能按时间随便删，否则会丢离线消息

新线程模型

为什么可以进 50ms

Partager cette Q&R