This article deeply explores the core mechanism of Redis Stream after confirmation: Pending queue (PEL) management, dead letter processing and exception recovery. Focus on analyzing PEL’s two-layer structure, state flow, memory overhead and four hard boundaries (cluster, copy, persistence, elimination), emphasizing that “at least once” relies on business idempotence, the number of deliveries ≠ the number of failures, and ghost entries need to be actively cleaned up. Combined with version evolution (7.0+ optimization), an automatic takeover strategy and dead letter flow design based on idle timeout are proposed, revealing the engineering complexity of Stream as a reliable MQ – normal paths are simple, and abnormal paths require meticulous operation and maintenance.
This article is a companion article to “Redis Stream Depth: Consumer Group, Pending Queue and Dead Letter Processing”. The previous article has finished talking about “how to consume, how to confirm, and how to create a group”; this article only talks about what happens after confirmation – how to recycle pending, how to retry to a dead letter if it fails, and whether Stream should be used as MQ. The command list, rax structure, position and bounding are retained in the previous article and will not be repeated in this article.
0.1 version node
Do not guess the read-write attributes and ACL classification based on the command name:XACKDEL / XNACK Such commands that can write PEL are write commands.XPENDING / XINFO Belongs to reading. There is an “OR” relationship between ACL categories, only for accounts @write cannot be executed when it belongs to @stream of XNACK。
measured boundary: Behavior of 7.x (XAUTOCLAIM three sections,XPENDING four forms,XCLAIM JUSTID、XGROUP DELCONSUMER、XTRIM With PEL,XINFO CONSUMERS Fields and Lua in the following two paragraphs) were all verified by actual running on Redis 7.4.2; 8.2 / 8.4 / 8.6 / 8.8 / 8.10 were only verified according to official documents and official blogs, and were not tested on 8.x instances. Before going online, use the target version of COMMAND DOCS The actual return shall prevail.
The reason why these items are listed separately is because they all changed the semantics of “after confirmation”: before 7.0, ghost entries could only be fished out manually; before 7.2, it was impossible to distinguish between “the consumer hung up” and “the consumer just had no work for the time being”; before 8.2, a two-step operation was required; before 8.4, the inspection cost increased linearly with the number of pending items; before 8.10, the total amount of multi-stream reads could not be controlled at all. The version difference is one year, and the operation and maintenance workload is doubled.
redis in code block # The line is a comment, but redis-cli Not recognized when reading stdin through pipe #(report ERR unknown command '#'), filter out the comment lines before executing the entire section:grep -v '^[[:space:]]*#' cmds.txt | redis-cli ...。
0.2 Four terms
entries-read: The logical sequence number of the group’s delivery. As the delivery progresses, it will also beSETIDrewrite.lag：entries-added - entries-readThe estimation difference is,XINFO GROUPSmay be NULL.delivery-count: This message has been delivered several times, only in PEL, not equal to the number of business failures.idle: The number of milliseconds from the last delivery or claim to now. It is only the timestamp difference, not a timeout.
0.3 Division of labor with the previous article
The previous article is responsible for “how a message is read and confirmed”; this article is responsible for “what to do if it gets stuck before confirmation”. The stream names and consumer names in the two articles can be compared and the codes can be spliced and used.
There is only one thing between “Stream can be used as a queue” and “Stream can be used as a reliable queue”: whether unacknowledged messages will disappear on their own. PEL is that account, the account is not balanced, and all previous “at least once” promises are empty words.
Three places where it is easiest to make preconceptions:
- Once you read it, you will consume it.. The read only moves the message into PEL. The business is not executed or confirmed, and the message is still alive. After the process is restarted, it is still there, but no one gets it back.
- If there are too many pendings, consumption will be slow.. It may be that the consumer has crashed, is stuck in an infinite loop, or the message has been
XTRIMThe main text was cropped and became a ghost entry – the three reasons were treated in completely different ways. - The high number of deliveries means there is something wrong with this message.。
delivery-countIt will be due to inspection, claiming,FORCEThere is no one-to-one correspondence between the rise and the failure of the business.
These three areas fall into specific items. The anti-pattern table in Chapter 6 is listed one by one according to “phenomenon + root cause + avoidance”. Here we only do positioning.
Why is there the impression that “Stream is lightweight MQ”? Because the normal path does look like:XADD Write,XREADGROUP read,XACK Confirm, four lines of commands will work. Watersheds only occur along abnormal paths—consumers crash, messages get stuck, and text gets cut. The simplicity of the normal path masks the entire engineering effort of the abnormal path.
2.1 Two-layer PEL and four fields
Redis has group-level PEL (recording unconfirmed messages in each group) and consumer-level PEL (recording who these messages belong to). Both layers point to the same batch of message IDs, but deleting either layer does not mean that the other layer is cleared simultaneously, so you need to distinguish which layer to look at when troubleshooting.
XPENDING Each row of the details has four fields: message ID, consumer to which it belongs,idle millisecond,delivery-count. These four fields determine all subsequent judgments – look at idle for timeouts, count for poison messages, and consumer names for ownership changes.
When troubleshooting, first determine the layer:XPENDING Give group level overview,XPENDING Details given to a certain consumer,XINFO STREAM Spread both layers together. If the three figures do not match, it usually means that some consumers have been DELCONSUMER Deleted – that batch of messages is no longer in any layer.
2.2 Status transfer
XADD → 进入流，未被任何组读
XREADGROUP ">" → 投递给某消费者，进 PEL，delivery-count = 1
处理成功 + XACK → 出 PEL；正文仍在流里，等 XTRIM / XACKDEL 回收
处理失败 / 崩溃 → 留在 PEL，idle 持续增长
idle 超阈值 → XCLAIM / XAUTOCLAIM / 8.4 CLAIM 转移归属，计数 +1
超重试上限 → 先写死信流，再 XACK 原消息
正文被删但 PEL 在 → 幽灵 pending；XAUTOCLAIM 第三段（7.0+）自动清
2.3 Impact 1: At least once, idempotence is the premise
XREADGROUP The semantics are at least once: crash, claim, and repeated delivery will cause the same message to be executed twice. Therefore, business side effects must have their own idempotent keys. The easiest thing to do is to use the business primary key (order number + event ID) to build a unique index instead of counting on “Stream guaranteed to be executed only once.”
There are two options for idempotent keys, with different trade-offs: write-only business primary key (order number) is the most stable, but different events of the same order will be treated as duplicates; write-only event IDs are more detailed, but require the production end to generate stably and remain unchanged during replay. Most systems use a combination of “business primary key + event type” to hand over uniqueness to the database’s unique index. Redis’s deduplication key is only used as a quick short-circuit.
2.4 Impact 2: PEL is an independent memory overhead and will not shrink over time.
The PEL entry remains in memory until it is ACKed; it does not become smaller due to the passage of time or repeated deliveries.only XACK and XACK DEL The series made it smaller. There are three growth paths: consumption rate is lower than production, abnormal branches miss writing ACK, and dead letters are not processed, causing failure messages to hang permanently. It also slows down patrol scans, so “PEL total” should be alerted more than “flow length”.
There are two calibers of single overhead, don’t mix them:
The latter comes from the actual measurement of the difference method on the official blog: after inserting 200,000 items and reading PEL, the memory increases from 6.80 MB to 41.53 MB, an increment of 34.73 MB, which is equivalent to about 174 B/item; the time index of 8.4 plus 3.55 MB (3.55 / 41.53 about 8.5%), the same caliber is about 17.8 B/item. Making a budget of only 30 to 50 B will underestimate the memory by 3 to 5 times, and a single consumer failure can overwhelm the memory.
What does the time index in 8.4 bring in exchange for this: Before, each delivery had to find the record along the PEL linked list and then change the time. The pending time was O(n); after adding the index, positioning by time was close to O(1). The price is 17.8 B more pending and one more maintenance on the write path – a typical transaction of “exchanging memory for CPU”. It also changes the “search for virus messages sorted by idle” from a full table scan to a range query.
2.5 Impact 3: A set of “visibility timeout”, but you have to set the threshold yourself
XPENDING Look at idle,XCLAIM according to min-idle-time Claim, combined is equivalent to SQS visibility timeout. but idle is not a timeout: When Redis reaches the point, it will not be automatically declared dead or re-rolled. The timeout is entirely a business-defined threshold.
The lower limit should be greater than the normal processing time of p99 plus ACK round trip, and experience is 3 to 5 times the normal time (5 to 10 s for 300 ms processing, and about 90 s for 30 s processing). The upper limit is the longest recovery delay that can be tolerated; if N retries are allowed, the wait for the k-th retry is approximately k × min-idle-timeand infer accordingly.
There are two typical failures when the threshold is set incorrectly. The setting is too short: the message is snatched away while it is still being processed, the same event is executed twice, and duplication can be seen in the log but the error rate indicator does not change. If the setting is too long: If consumers are really broken, they will have to wait a long time for someone to take over, which is manifested as “the backlog does not decrease and idle keeps rising.” Neither of them is reflected as an error, it can only be seen from the idle distribution.
2.6 Impact 4: Delivery count is not equal to the number of business failures
The conclusion is straightforward:The retry criterion uses business failure count, and the delivery count only serves as a hard upper limit.. Tested both –XPENDING Read three times in a row, the count is still 1;IDLE Filtering returns idle to zero but does not change the count, and it only filters but does not sort, and returns are still in ascending order by ID.
2.7 Impact 5: Ghost pending has two reading paths
Text quilt XDEL or XTRIM Deleted and PEL is still there, it becomes a ghost:XPENDING can still see it,XREADGROUP When reading the specific ID, the text is empty, and the delivery count still increases. There are two ways to clear it——XAUTOCLAIM The third paragraph return will delete it from PEL (7.0+), or scan it yourself DeletedIds Supplementary XACK。XTRIM Only the text is moved but the PEL is not, so “cropping” itself is an act of creating ghosts.
The most trouble-free identification method is to compare two commands:XPENDING A certain ID can be found inXRANGE But an empty array is returned – this combination can basically conclude that the text was deleted first. If the inspection sees it, do not retry the service, just clear the PEL and alert.
2.8 Impact Six: Four Hard Boundaries of Cluster, Replica and Persistence
- cluster: Stream key will not be automatically sharded, large keys only fall into a single shard; or business sharding (
orders: delivered >= HardCeiling)), or move out in advance.
return await MoveToDeadAsync(entry, businessFailures, ex);// 留在 PEL，等 Janitor 认领重试
return false;
- copy: The master-slave copy is the command stream.
XAUTOCLAIMThe scan cannot be offloaded for you on the slave database – the read extension is not effective for PEL inspection. - Endurance：AOF
everysecAt most 1 second of writes will be lost, but that 1 second may containXACK“Confirmation is lost” is more difficult to check than “Message is lost”. - Memory obsolescence：
maxmemory-policyOnce elimination is triggered, the Stream may be deleted by the entire key, and the PEL disappears and cannot be recovered; pending streams must be configurednoevictionor standalone instance.
What these four have in common is thatNeither will be exposed on normal paths: Cluster imbalance is only manifested by a certain shard memory being full first; copy delay is only manifested by a small amount of replay after failover; the 1 second lost by AOF only appears at the moment of the crash; elimination is the silent deletion of keys. All four need to be configured in advance and cannot be made up afterwards.
2.9 Irreversible operations
XGROUP DELCONSUMER It will be discarded together with the unACKed messages in the name of the consumer and deleted. XPENDING return 0 / nil / nil / nilthe message can never be claimed again;XGROUP SETID / XSETID Changing the location will cause extensive replay or permanent message loss. These two types of operations are either double-checked or not touched at all. Please create a new group for replay.
2.10 Eight-element skeleton: take a look for yourself
DEL demo:pending
XADD demo:pending 1000-0 orderId A1 status created
XGROUP CREATE demo:pending g 0
XREADGROUP GROUP g w1 COUNT 1 STREAMS demo:pending >
XPENDING demo:pending g - + 10
XCLAIM demo:pending g w2 0 1000-0
XPENDING demo:pending g - + 10
XACK demo:pending g 1000-0
The expected order is:1 → 1000-0 → OK → One delivery → 1000-0 / w1 / idle 1 / 1 → 1000-0 → 1000-0 / w2 / idle 1 / 2 → 1. idle is millisecond-level timing, and the actual measured value is 1 (the random machine and execution speed may be 0 or 1). The following text will be written according to the measured value. The count in line 7 changes from 1 to 2, which is direct evidence that “claiming will increase the count”; both lines 5 and 7 are XPENDINGthe count remains unchanged, which is direct evidence that “query does not change the count”. Do not write comments in this block so that the entire pipeline can be executed.
3.1 Four views on XPENDING
# 概要：总数 / 最小最大 ID / 按消费者分布
XPENDING mystream mygroup
# 明细（全组）：- + 是 ID 范围，10 是条数上限
XPENDING mystream mygroup - + 10
# 明细（按消费者）：消费者名只能追加在最后
XPENDING mystream mygroup - + 10 consumer-A
# IDLE 过滤：只筛 idle 大于 60000 ms 的条目，不排序、不改计数
XPENDING mystream mygroup IDLE 60000 - + 10
The four forms do not change the delivery count, which is the most easily misjudged point. If you want to search for poisonous information by idle, use the fourth method; if you want to confirm who a certain message belongs to, use the third method.
The summary form is most easily misused as “backlog monitoring”: it gives the total number of unconfirmed messages, excluding messages that have not been read by any consumers. The real backlog needs to be lag Looking at it together, focusing only on PEL will misjudge “production too fast” as “consumption too slow”.
3.2 XCLAIM and XAUTOCLAIM
# 精确认领：min-idle-time 毫秒；不加 JUSTID 计数 +1，加了不递增
XCLAIM mystream mygroup consumer-B 60000 1526569498055-4
# 自动认领：从游标 0-0 扫一批，返回 下一游标 / 认领列表 / 幽灵 ID 列表
XAUTOCLAIM mystream mygroup consumer-B 60000 0-0 COUNT 10
XCLAIM It’s a single item and you need to know the ID;XAUTOCLAIM It’s in batches and gives you a cursor to scan down by yourself. Use the latter for production inspection and the former for retrying a single item.COUNT The scanning limit is ×10, don’t expect COUNT 10 Just scan 10 items.
XCLAIM There are also two repair parameters:FORCE It is allowed to claim IDs that are not in PEL, which can be used to correct inconsistencies between PEL and the text;LASTID Put the group last-delivered-id Advance together. They are all repair methods. Do not touch the normal retry path.
3.3 8.2: Confirm and delete in one step
# 语法：三种引用选项三选一（缺省 KEEPREF），IDS 后必须先写条数
XACKDEL key group [KEEPREF | DELREF | ACKED] IDS numids id [id ...]
XDELEX key [KEEPREF | DELREF | ACKED] IDS numids id [id ...]
7.x can only XACK + XDEL Take two steps, and if it collapses in the middle, the ghost will be left; starting from 8.2, one command will be completed atomically.IDS numids inside numids Must be written, this is the most commonly missed parameter.
The return value expresses “how much was deleted”, not “success”: no error is reported when some IDs do not exist, but less is deleted. Use multiple sets of scenes first XPENDING Confirm the status of each group before selecting options, otherwise it is easy to have the illusion of “thinking it has been deleted, but the text is actually still there.”
3.4 CLAIM of 8.4 and upper limit of 8.10
# 8.4：一次读顺带认领 idle 超阈的旧消息，省掉一轮单独巡检
XREADGROUP GROUP g c COUNT 100 BLOCK 5000 CLAIM 30000 STREAMS s >
# 8.10：COUNT 是每流上限，MAXCOUNT 才是跨流累计上限
XREADGROUP GROUP g c COUNT 50 MAXCOUNT 80 STREAMS s1 s2 s3 > > >
# 8.10：MAXSIZE 按字节封顶（跨流累计）
XREADGROUP GROUP g c MAXCOUNT 5 MAXSIZE 100000 STREAMS s1 s2 s3 > > >
Official benchmark CLAIM The income is quantified at 22.5 times: 200,000 pending, 1,000 idle,COUNT=1000repeated 1000 times, a single round trip is 54.671 ms vs. 2.426 ms. This multiple is not a universal constant – it depends on the time in a scanThe ratio of the number n of items that can actually be claimed to the upper limit of scanning kthe smaller n/k is, the fewer round trips are saved. When the quantity and version don’t match up, don’t take it as a promise.
Got it CLAIM Is there no need for Janitor? Lightweight scenarios of single instance and single stream can be omitted; as long as multiple instances or independent alarms are involved, Janitor still needs to have——CLAIM It only addresses “incidentally claiming”, not “who is responsible for monitoring and reporting”.
3.5 8.8 XNACK
# SILENT：认作可重试，投递计数 -1，归属不变
XNACK orders:events order-group SILENT IDS 1 1750000000000-0
# FAIL：标记失败，计数不变，优先于普通待投递消息
XNACK orders:events order-group FAIL IDS 2 1750000000001-0 1750000000002-0
# FATAL：进「不可再投递」区，idle 置 LLONG_MAX
XNACK orders:events order-group FATAL IDS 1 1750000000003-0
Its value is to write “this item was not processed” back to PEL, rather than relying on business-side counting inference. There are only @streamonly awarded @write The account cannot be executed.
Don’t think of it as a “retry switch”:SILENT Just decrementing the count by 1 will not make the message deliverable in advance; when it can be re-delivered is still determined by min-idle-time Decide.
3.6 .NET signature and six pitfalls
Four of the six pitfalls are “parameter names or types do not match”, and two are “semantic understanding deviations”, all of which are types that can be encountered during compilation or the first time it goes online. Pasting this table into the project README is faster than reading the official XML.
Pit 4 is the most typical: when accurately locating by ID in the text,StreamPendingMessagesAsync(_stream, _group, 1, RedisValue.Null, minId: id, maxId: id)——If the fourth parameter is omitted, the compilation will fail directly (2.8.x reports CS7036,3.x report CS1501）。
4.1 Three layers
The main loop manages “reading” and “processing”, the Janitor manages “timeout recycling”, and the dead letter stream manages “abandonment”. The three share the same set of idempotence and failure counts, and any layer added alone will miss the mark. Without Janitor, stuck messages will never get a second chance; without dead letters, retries will turn into infinite loops; without unified idempotence, the first two will create repeated executions.
4.2 Main loop
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using StackExchange.Redis;
public sealed class OrderEventConsumer
{
private readonly IDatabase _db;
// 业务失败计数：唯一的重试判据，与投递计数是两套账
private readonly IFailureCounter _failures;
private readonly Func _handle;
private readonly RedisKey _stream = "orders:events";
private readonly RedisValue _group = "order-group";
private readonly RedisKey _dead = "orders:dead";
private readonly string _consumer = $"w- delivered >= HardCeiling)
return await MoveToDeadAsync(entry, businessFailures, ex);
// 留在 PEL，等 Janitor 认领重试
return false;
-
// 查 PEL 只按 ID，不限消费者：认领后归属会变成 janitor，
// 若按主循环消费者名过滤会查不到，delivered 恒为 1，毒消息无限重试
var detail = await _db.StreamPendingMessagesAsync(
_stream, _group, 1, RedisValue.Null, minId: entry.Id, maxId: entry.Id);
long delivered = detail.Length == 0 ? 1 : detail[0].DeliveryCount;
// 业务失败次数：真正的重试判据
int businessFailures = await _failures.IncrementAsync(eventId);
// 两条阈值任一命中即转死信
if (businessFailures >= MaxBusinessFailures ";
private const int BatchSize = 10;
private const int MaxBusinessFailures = 5;
// 投递计数只兜底，不参与正常重试判定
private const int HardCeiling = 20;
public OrderEventConsumer(IConnectionMultiplexer redis, IFailureCounter failures,
Func handle)
public async Task RunAsync(CancellationToken ct)
{
await EnsureGroupAsync();
while (!ct.IsCancellationRequested)
{
try
{
// 不带 BLOCK：StackExchange.Redis 没有 blockMilliseconds 参数
var entries = await _db.StreamReadGroupAsync(
_stream, _group, _consumer,
StreamPosition.NewMessages, count: BatchSize);
if (entries.Length == 0)
{
// 空结果退避
await Task.Delay(200, ct);
continue;
}
foreach (var entry in entries)
await ProcessSafelyAsync(entry, ct);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; }
catch (RedisException ex)
{
// Redis 不可用：退避重试，别让进程崩
Console.WriteLine($"Redis 异常：{ex.Message}");
await Task.Delay(1000, ct);
}
}
}
private async Task EnsureGroupAsync()
{
try
{
// 建组位点 $：只消费建组之后的新消息
await _db.StreamCreateConsumerGroupAsync(
_stream, _group, StreamPosition.NewMessages, createStream: true);
}
catch (RedisServerException ex) when (ex.Message.Contains("BUSYGROUP"))
{
// 组已存在：正常，不重建、不重置位点
}
}
// 返回 true 表示已收口：成功 ACK，或已写死信并 ACK
public async Task ProcessSafelyAsync(StreamEntry entry, CancellationToken ct)
{
// 幽灵条目：正文被裁掉，只清 PEL 并记审计
if (entry.Values == null || entry.Values.Length == 0)
{
await _db.StreamAcknowledgeAsync(_stream, _group, entry.Id);
return true;
}
// 业务唯一事件 ID：重复投递时靠它做幂等
var eventId = entry["eventId"];
// 订单号与事件类型只用于业务处理，不参与幂等
var orderId = entry["orderId"];
var eventType = entry["type"];
// 兜底：没带 eventId 就用 Stream ID
if (eventId == RedisValue.Null) eventId = entry.Id;
try
{
await _handle(eventId, orderId, eventType, ct);
await _db.StreamAcknowledgeAsync(_stream, _group, entry.Id);
// 成功即清零，避免累计值误杀后续正常消息
await _failures.ResetAsync(eventId);
return true;
}
catch (Exception ex)
{
// 查 PEL 只按 ID，不限消费者：认领后归属会变成 janitor，
// 若按主循环消费者名过滤会查不到，delivered 恒为 1，毒消息无限重试
var detail = await _db.StreamPendingMessagesAsync(
_stream, _group, 1, RedisValue.Null, minId: entry.Id, maxId: entry.Id);
long delivered = detail.Length == 0 ? 1 : detail[0].DeliveryCount;
// 业务失败次数：真正的重试判据
int businessFailures = await _failures.IncrementAsync(eventId);
// 两条阈值任一命中即转死信
if (businessFailures >= MaxBusinessFailures || delivered >= HardCeiling)
return await MoveToDeadAsync(entry, businessFailures, ex);
// 留在 PEL，等 Janitor 认领重试
return false;
}
}
private async Task MoveToDeadAsync(StreamEntry entry, int failures, Exception ex)
{
// 顺序：先写死信，成功后才 ACK 原消息；反了会丢一条
await _db.StreamAddAsync(_dead, new NameValueEntry[]
{
new("originalId", entry.Id.ToString()),
new("payload", string.Join(";", entry.Values.Select(v => $"{v.Name}={v.Value}"))),
new("error", ex.Message),
new("failures", failures.ToString()),
new("deadAt", DateTimeOffset.UtcNow.ToString("O"))
}, maxLength: 100000, useApproximateMaxLength: true);
await _db.StreamAcknowledgeAsync(_stream, _group, entry.Id);
return true;
}
}
public interface IFailureCounter
{
Task IncrementAsync(RedisValue eventId);
Task ResetAsync(RedisValue eventId);
}
// 最小可用实现：String 计数 + TTL，成功时删键清零
// 强一致场景换成数据库幂等表，与业务结果同一事务提交
public sealed class RedisFailureCounter : IFailureCounter
{
private readonly IDatabase _db;
private static readonly TimeSpan Ttl = TimeSpan.FromDays(7);
public RedisFailureCounter(IConnectionMultiplexer redis) => _db = redis.GetDatabase();
public async Task IncrementAsync(RedisValue eventId)
{
// 按事件 ID 计数，不用「订单号 + 事件类型」
RedisKey k = "fail:" + eventId.ToString();
long n = await _db.StringIncrementAsync(k);
// 每次失败续期，TTL 要覆盖最长重试窗口
await _db.KeyExpireAsync(k, Ttl);
return (int)Math.Min(n, int.MaxValue);
}
public async Task ResetAsync(RedisValue eventId)
{
// 成功即删键：下次失败从 1 重新计
await _db.KeyDeleteAsync("fail:" + eventId.ToString());
}
}
Three points are worth mentioning individually. First, for group building $ instead of 0:use 0 All historical messages will be replayed when restarting. The correct way is to build $ Rely on Janitor to recycle history (0 For intentional replay only). Second, the PEL check must not be limited to consumers, otherwise the messages claimed by Janitor cannot be found, and the count will always be 1. Third, the order of dead letters and ACK cannot be reversed.
There is another thing that is easily overlooked:BatchSize Don’t go big. The throughput of taking 10 items at a time is not much lower than taking 100 items. However, if a single batch is too large, the idle distribution will be distorted – the items at the end of the queue have not started to be processed, and the timer is already running. Inspection will treat them as “stuck” and snatch them away in advance.
4.3 Janitor
using System;
using System.Threading;
using System.Threading.Tasks;
using StackExchange.Redis;
public sealed class PendingJanitor
{
private readonly IDatabase _db;
private readonly Func> _handle;
// 告警上报：幽灵 ID 必须进监控系统，不能只打印
private readonly Func _alert;
// 单实例租约：Redis 锁、etcd 或部署平台都行
private readonly ILease _lease;
private readonly RedisKey _stream = "orders:events";
private readonly RedisValue _group = "order-group";
private readonly string _consumer = $"janitor-{Environment.MachineName}-{Environment.ProcessId}";
private const long MinIdleMs = 60_000;
private const int BatchSize = 50;
public PendingJanitor(IConnectionMultiplexer redis,
Func> handle,
ILease lease, Func alert)
{
_db = redis.GetDatabase();
_handle = handle;
_lease = lease;
_alert = alert;
}
public async Task RunAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
try
{
// 只有拿到租约的实例巡检，否则多实例互相抢同一批消息
if (await _lease.TryAcquireAsync("janitor:orders:events", ct))
await SweepOnceAsync(ct);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; }
catch (Exception ex)
{
// 巡检异常不能终止常驻任务
Console.WriteLine($"巡检异常：{ex}");
}
// 巡检周期：太密抢 CPU，太疏拉长恢复时间
await Task.Delay(30_000, ct);
}
}
private async Task SweepOnceAsync(CancellationToken ct)
{
// 游标从 0-0 起扫，回到 0-0 表示一轮结束
RedisValue cursor = "0-0";
do
{
var result = await _db.StreamAutoClaimAsync(
_stream, _group, _consumer,
minIdleTimeInMs: MinIdleMs, startAtId: cursor, count: BatchSize);
// 必须用返回值推进游标，写死 0-0 会一直扫同一批
cursor = result.NextStartId;
foreach (var entry in result.ClaimedEntries)
await _handle(entry, ct);
// 7.0+：正文已裁掉的幽灵 ID，XAUTOCLAIM 已从 PEL 清掉
foreach (var ghost in result.DeletedIds)
{
Console.WriteLine($"幽灵 ID：{ghost}");
await _alert($"stream.ghost:{ghost}");
}
} while (cursor != "0-0" && !ct.IsCancellationRequested);
}
}
public interface ILease
{
Task TryAcquireAsync(string name, CancellationToken ct);
}
SweepOnceAsync There are two details that must be kept: the cursor is advanced with the return value;DeletedIds If it is not empty, it means that some messages have been clipped and eaten. This is a warning signal of “MAXLEN is set too small” and must be reported instead of discarded. Janitor is shared with the main loop ProcessSafelyAsyncso there will not be two sets of dead letter judgments.
4.4 Five rules for closed dead letter loops
- Write first then delete: The original message will be ACKed only when the dead letter is successfully written. If the order is reversed, one message will be lost and the message will crash.
- write full context: Original ID, complete fields, error information, number of failures, timestamp. Just writing the ID is equivalent to pushing the troubleshooting work to the future.
- The flow of dead letters also needs to be capped：
MAXLEN ~ 100000otherwise the dead letter itself will become a new memory bomb. - Dead letters must have a consumer: If you write it, no one will read it, it just postpones the loss of the message; the consumption of dead letter flow must be able to alert, replay, and manually take over.
- Transfers must also be idempotent: The same message may be written to the dead letter by two instances at the same time, and the original ID is used as the deduplication key.
Field design is more important than command: at least the original ID, original field, error, number of failures, and writing time. Without the original field, you can only rely on guessing during replay; without the number of failures, you can’t tell whether it “exploded the first time” or “retryed 20 times before exploding.”
4.5 Safely delete consumers
Before deleting, you must first confirm that there is no pending message, otherwise the unconfirmed message will disappear and cannot be claimed. Checking and deleting should be atomic:
-- KEYS[1] = stream key；ARGV[1] = group；ARGV[2] = consumer
-- 返回：1 已安全删除 / -2 仍有 pending / -1 key 不存在
if redis.call('EXISTS', KEYS[1]) == 0 then
return -1
end
local pending = redis.call('XPENDING', KEYS[1], ARGV[1], '-', '+', 1, ARGV[2])
if type(pending) == 'table' and #pending > 0 then
return -2
end
redis.call('XGROUP', 'DELCONSUMER', KEYS[1], ARGV[1], ARGV[2])
return 1
7.4.2 The above three inputs are returned respectively -2 / 1 / -1consistent with the comments.
4.6 Idempotence and order preservation by business primary key
Deduplication key (SET ack:{eventId} 1 NX EX 86400) can only be used as a quick short-circuit. Once the TTL arrives and the message is claimed, it cannot prevent duplication; true idempotence relies on the unique index of the database, and the setting of the deduplication key and the business writing must be in the same transaction semantics, and ACK will only occur after the business confirmation is successful.
In terms of order preservation, Stream itself only guarantees ordering by ID within the same stream, and does not guarantee the processing order across consumers. To preserve order by order, you can only use “same primary key routing to the same consumer” to add consumer serialization; cross-process (Janitor claim, expansion) still depends on the business version number, so deduplication, version verification, and primary key routing usually need to be done together.
The specific method of disclosing the version number: the message body carries the business version (seq or updatedAt), leave when leaving the warehouse WHERE seq > 已存 seq Conditional updates, repeated and out-of-order writes are naturally rejected, and no additional locks are required.
4.7 Capacity estimation
单条消息 = 固定开销(约 40 至 80 B：rax 分摊 + 16 B ID + listpack 元数据)
+ sum(字段名 + 值 + 每项约 1 至 3 B 编码)
整 key = 单条 × 条数
+ 各消费组 × (约 100 B + 消费者数 × 常数)
+ 未确认条数 × 约 174 B(PEL 端到端)
+ 8.4 时间索引(约 17.8 B/条 pending)
+ 余量 1.2 至 1.5 倍(复制缓冲、碎片、dict 扩容)
MAXLEN >= 峰值写入速率 × 最大消费延迟 × 安全系数
The three sets of calculations only give magnitude reference, and the actual values must be replaced after self-test according to the method in 4.8. 200,000 200B messages, 10,000 pending: the main text is about 40 MB, the PEL is about 1.7 MB, and the time index is about 0.18 MB; when the pending is raised to 200,000, the PEL alone becomes about 35 MB, which is larger than the main text. This is why Section 2.4 emphasizes dual calibers.
Let’s give another group that is easily missed: 10,000 100B messages, hung on 10 consumer groups, each group has 10,000 pending. The main text is about 1 MB, each PEL is about 1.7 MB, and the ten groups total about 17 MB – the PEL is 17 times larger than the main text. When multiple groups share a flow, the cost is linearly superimposed according to the number of groups, which is the most easily missing item in the budget.
4.8 Monitoring and self-testing
Use the difference method for self-assessment, don’t believe it MEMORY USAGE——It does not count as PEL:
# 基线：造数前取一次，不给基线就没有差值
redis-cli INFO memory | grep used_memory:
# 造数后：与基线相减 = 正文增量
redis-cli INFO memory | grep used_memory:
# 投递后：与上一次相减 = PEL 增量（读 10 万条不 ACK）
redis-cli INFO memory | grep used_memory:
# 单次调用平均耗时：只看 XAUTOCLAIM 这一轮
redis-cli INFO commandstats | grep xautoclaim
# 尾部延迟：平均值会掩盖长尾
redis-cli SLOWLOG GET 10
lag is NULL, don’t panic, and don’t SETID. One of the causes of NULL is that the group position has been changed, and it is transient: once the group delivers the last message in the stream to the member,entries-read will be reset to the correct value,lag Automatic recovery. During the NULL period, press “PEL total + maximum idle + XLEN “Growth” estimated backlog, and the alarm was downgraded to observation. The collection frequency does not need to be high: the total number of PELs is once every 1 minute, the maximum idle time is once every 5 minutes,lag and XLEN Just pick them together. The inspection script itself also needs to limit the current flow.XPENDING Details are heavy commands on large PEL, and brushing by seconds will turn inspection into a new load source.
Two more things to remember: just look XLEN Will be leaked (after all the messages are delivered) lag It can be 0,XLEN Stable, but PEL is rising); just looking at the average will miss the tail (normal total volume does not mean that there are not a few idle and extremely poisonous news).
5.1 Delay queue: ZSet trigger + Stream consumption
ZSet stores the “expiration time”, and after the expiration date, it is atomically taken out and written into the Stream for consumption. The atomicity of extraction is guaranteed by a piece of Lua:
-- KEYS[1] = delay:{order}（ZSet）；KEYS[2] = events:{order}（目标 Stream）
-- ARGV[1] = now(ms)，ARGV[2] = batch
-- Cluster 下两个 key 必须同 slot：用同一个 hash tag {order} 绑定
local due = redis.call('ZRANGE', KEYS[1], '-inf', ARGV[1], 'BYSCORE', 'LIMIT', 0, ARGV[2])
for i, member in ipairs(due) do
redis.call('ZREM', KEYS[1], member)
redis.call('XADD', KEYS[2], 'MAXLEN', '~', 200000, '*', 'orderId', member)
end
return #due
7.4.2 Return when two items have expired and one item has not yet expired. 2，ZCARD There are 1 left, and there are exactly two in the Stream – retrieval and deletion in the same script, so there will be no repeated triggering of “retrieval but not deletion”.
5.2 Conclusions on the three topics
- Event sourcing: One business fact and one message. The projection function must be a pure function.
XRANGE - +full replay,XRANGE -Fixed-point replay can reproduce the same state. - message body: Objects exceeding several KB are placed in external storage, and only references are placed in the Stream; large messages will push up the rax node and the copy buffer.
- IoT telemetry: Sharded by device hash (
device:events:{sh}), idempotent keys are useddev:tsthe deduplication key only makes a short circuit, and the unique index takes care of everything.
The commonality of the three topics is: the message is just the carrier, and the real cost lies in “how to replay and how to remove duplicates.” Think clearly about the replay path first, and then decide what should be stored in the message – if you store external references, you must ensure that the references can be retrieved in the long term, and if you store the full amount, you must accept the memory cost.
5.3 Selection matrix
5.4 Three counter-intuitive truths
- What you save is MQ, but what you pay is the engineering quantity.. Stream does not require new components, but the five things of inspection, dead letter, idempotence, capacity, and monitoring must be written by yourself. In calculation, it is “one MQ replaced by one set of code.”
COUNTIt’s the upper limit for each stream, not the total amount.. Used when reading multiple streams simultaneouslyCOUNTThe total amount of control will exceed expectations, 8.10MAXCOUNTIt is the cross-stream accumulation.XACKDELNot a panacea. When multiple groups share the same stream,KEEPREFTo prevent the text from being deleted, you must useACKEDOr wait for all groups to confirm; only dare to use it in single group scenarios.DELREF。
The three together form one sentence: Stream’s sweet spot is at the intersection of “existing Redis, medium volume, and willing to build its own operation and maintenance software.” If you leave this intersection, you will have to pay tuition.
5.5 Decision tree
消息能不能丢？
├─ 能，只要解耦或在线广播 ──► List / Pub/Sub
└─ 不能
├─ 需要「到未来某时刻才可见」？ ──► ZSet + Stream（5.1）
└─ 需要确认 / 重试 / 回溯？
├─ 长期保留、任意 offset、TB 级堆积 ──► Kafka
├─ JMS 语义、企业事务、原生 DLQ ──► ActiveMQ / Artemis
├─ 精确一次且不愿做业务幂等 ──► Kafka 事务或数据库事件表
└─ 已有 Redis、量中等、能自建幂等与巡检 ──► Redis Stream
5.6 When to move
Three signals: the memory of a single key exceeds half of the instance budget; the time taken for a single pending inspection enters the second level; the business begins to require precise one-time or multi-day playback. The migration path is double-write + site alignment: the new system first reads only without deleting, waits for the backlog on the Stream side to return to zero, then cuts the write, and finally uses the original ID to check the consistency.
The rollback plan must be written in advance: after cutting the stream, keep the read-only window on the Stream side for at least one inspection cycle, and confirm that there are no omissions in the new system before shutting down writing. Without a read-only window, rolling back is like gambling again.
6.1 Twelve anti-patterns
6.2 Eighteen-item online list
The list is in the order of “Configuration → Code → Operation and Maintenance”: the first 4 items do not need to be read down, they are collapsed; the middle 8 items determine whether the abnormal path can be passed through; the last 6 items determine whether you can see the fault.
DEL demo:pending
XADD demo:pending 1000-0 orderId A1 status created
XGROUP CREATE demo:pending g 0
XREADGROUP GROUP g w1 COUNT 1 STREAMS demo:pending >
XPENDING demo:pending g - + 10
XCLAIM demo:pending g w2 0 1000-0
XPENDING demo:pending g - + 10
XACK demo:pending g 1000-0
XPENDING demo:pending g
XLEN demo:pending
XADD demo:pending 1001-0 orderId A2 status paid
XREADGROUP GROUP g w1 COUNT 1 STREAMS demo:pending >
XDEL demo:pending 1001-0
XPENDING demo:pending g
XAUTOCLAIM demo:pending g w2 0 0-0 COUNT 10
XPENDING demo:pending g
XRANGE demo:pending - +
XADD demo:pending 1002-0 orderId A3 status done
XREADGROUP GROUP g w1 COUNT 1 STREAMS demo:pending >
XACK demo:pending g 1002-0
XDEL demo:pending 1002-0
XPENDING demo:pending g
XRANGE demo:pending - +
XINFO GROUPS demo:pending
DEL demo:pending
Expected (7.4.2 actual test is successful; scripts and tables correspond to line numbers, so no comments are written in the text):
