diff --git a/README.md b/README.md index 4bb92eb..fc793e5 100644 --- a/README.md +++ b/README.md @@ -222,18 +222,35 @@ public class OrderConsumer extends EnhanceStreamConsumerHandler { ## 七、配置项 -| 配置项 | 默认值 | 说明 | -|---|---|---| -| `redis-stream.env-isolation` | `false` | 是否按环境隔离 stream 名 | -| `redis-stream.env` | `""` | 环境标识,隔离开启且非空时作为 stream 名后缀(`topic_env`) | -| `redis-stream.consumer-name` | `""` | 消费者名;为空时自动生成唯一名(**见下方注意事项**) | -| `redis-stream.read-count` | `10` | 单次 `XREADGROUP` 拉取条数 | -| `redis-stream.block-ms` | `2000` | `XREADGROUP` 阻塞毫秒 | -| `redis-stream.max-retry` | `3` | **全局**最大投递次数(含首投),超限落死信;可被 handler 的 `getMaxRetryTimes()` 单独覆写 | -| `redis-stream.autoclaim.min-idle-ms` | `60000` | 消息 idle 多久后可被认领重投 | -| `redis-stream.autoclaim.interval-ms` | `30000` | autoclaim 扫描周期 | -| `redis-stream.trim.interval-ms` | `60000` | `XTRIM` 执行周期 | -| `redis-stream.trim.maxlen` | `100000` | 近似保留的最大 stream 长度(`MAXLEN ~`) | +> 全部可配项如下,前缀 `redis-stream`,**均为可选**——不配即用注释中的默认值。Spring 宽松绑定,`read-count` / `readCount` 等写法等价。 + +```yaml +redis-stream: + # ===== 基础 ===== + env-isolation: false # 是否按环境隔离 stream 名:开启后名字追加 _env 后缀,多环境共用一套 Redis 时互不串扰 + env: "" # 环境标识,env-isolation=true 且非空时作为 stream 名后缀(topic_env);收发双方须配一致 + consumer-name: "" # 消费者名,留空自动生成唯一名。多实例务必留空,写死会被当作同一消费者互相争抢、破坏负载均衡(见「八、注意事项」) + read-count: 10 # 单次 XREADGROUP 拉取条数(一次最多取多少条来处理) + block-ms: 2000 # XREADGROUP 阻塞等待毫秒:无新消息时阻塞这么久再返回,影响拉取实时性与空转开销 + max-retry: 3 # 全局最大投递次数(含首投),超限落死信;可被 handler 的 getMaxRetryTimes() 单独覆写 + + # ===== 崩溃恢复 · 自动认领:消费者宕机/超时未 ACK 的消息滞留 PEL,由后台定时认领重投(机制见「六、消费语义」) ===== + autoclaim: + min-idle-ms: 60000 # 消息 idle(距上次投递)超过此值(ms)才可被其它消费者认领重投。太小会与正常重试争抢,太大则故障恢复慢 + interval-ms: 30000 # autoclaim 扫描 PEL 的周期(ms) + + # ===== Stream 长度裁剪:防无界增长,后台定时 XTRIM MAXLEN (精确裁剪),保留最新 maxlen 条 ===== + trim: + maxlen: 100000 # 保留的最大条数(精确裁剪)。默认 10万;要 5万就设 50000,超出部分删最老条目 + interval-ms: 60000 # 裁剪执行周期(ms)。调小可让实际长度更贴近 maxlen(代价是 XTRIM 执行更频繁) +``` + +Stream 裁剪有两点**必须知道**: + +1. **软上限,非实时硬卡。** 两次裁剪之间 stream 会短暂超过 `maxlen`,峰值 ≈ `maxlen + 生产速率 × interval-ms`。例:`maxlen=50000`、每秒进 1000 条、周期 60s ⇒ 下次裁剪前最高可达 `11 万`,随后拉回 5 万。要更紧就调小 `interval-ms`。 +2. **按条数裁剪,不区分是否已消费。** `XTRIM MAXLEN` 只删最老条目,不管是否已 ACK。`maxlen` 设太小且消费积压超过它时,**未消费的老消息会被直接裁掉 = 丢消息**(被裁的 PEL 消息会被静默强制 ACK)。⚠️ `maxlen` 要留足:至少覆盖「峰值生产速率 × 消费者最长停机/滞后时间」。5 万 / 10 万对一般业务够用,单条消费慢或实例可能长停时需上调。 + +> 注:`spring-data-redis 2.3.x` 高级 API 仅支持精确裁剪,无近似裁剪(`MAXLEN ~`)重载,故走精确裁剪。 --- @@ -242,7 +259,7 @@ public class OrderConsumer extends EnhanceStreamConsumerHandler { - **多实例部署不要写死 `consumer-name`**。留空时框架会为每个实例生成唯一消费者名(`consumer-xxx` / `claimer-xxx`),符合消费组负载均衡模型。若多个实例配置了相同的固定 `consumer-name`,它们会被视为同一个消费者而互相争抢,破坏负载均衡 —— `consumer-name` 仅建议用于单实例或本地调试。 - **环境隔离**:多环境(dev/test/prod)共用一套 Redis 时,开启 `env-isolation` 并设置 `env`,各环境 stream 名互不串扰。**收发双方必须配置一致**,否则 topic 对不上。 - **压缩契约**:仅 `body` 字段可压缩。`compress=true` ⇒ `enc=gzip`、`body=base64(gzip(json))`;否则 `enc=plain`、`body` 为明文 json。消费侧仅按 `enc` 解码,与 topic 无关,可与 C++ 等异构端互通。 -- **Stream 裁剪**:框架按 `trim.maxlen` 近似裁剪 stream 防止无界增长。请确保 `maxlen` 足够大,避免重试中的消息在达到 `max-retry` 前被裁掉(被裁掉的 PEL 消息会被静默强制 ACK)。 +- **Stream 裁剪**:框架按 `trim.maxlen` 精确裁剪 stream 防止无界增长(详见「七、配置项 · Stream 长度裁剪」)。请确保 `maxlen` 足够大,避免重试中的消息在达到 `max-retry` 前被裁掉(被裁掉的 PEL 消息会被静默强制 ACK)。 - **无 handler 时容器空转**:未注册任何 `EnhanceStreamConsumerHandler` Bean 时,监听/认领/裁剪线程均不启动,仅 `RedisStreamEnhanceTemplate` 可用于纯发送场景。 ---