队列

Redis Streams 消费组把新条目分配给具名消费者，并记录尚未确认的投递。它不保证业务副作用恰好发生一次，不会阻止保留策略删除数据，也不会自行恢复被遗弃的任务。将其用作工作队列前，应区分组的读取位置、待处理条目与流里实际保留的数据。 读取位置与待处理任务不同 XGROUP CREATE jobs workers 0-0 从现有历史之前起步；使用 $ 则从当前末端之后开始。同组成员分担新投递，不同消费组拥有各自独立的投递状态。并发工作进程应使用不同的消费者名字。最后投递位置属于整个组，并不是每个消费者分别提交一个偏移。 未使用 CLAIM 或 NOACK 的普通 XREADGROUP 以 > 请求新条目，投递后将它们记入待处理列表 PEL。组的位置在新条目投递时就推进，发生在 XACK 之前。XPENDING 统计的是已经投递但尚未确认的条目，不是从来没有投递过的积压。因此，待处理计数为零时，组里仍可能有大量未读工作。 XACK 把指定 ID 从该组 PEL 移除，不删除流里的记录，也不推进最后投递位置。确认可以乱序完成，其他组依然可以读取保留下来的条目。XREADGROUP 使用 0-0 时读取的是这个具名消费者自己的待处理历史，不是在请求新工作，也不会把另一位消费者名下的待处理消息交给它。 恢复需要显式决策 反复执行普通 > 读取，不会重试被遗弃的 PEL 条目。XAUTOCLAIM 能把满足条件的待处理条目转给请求方，参数必须包含流、组、消费者、最小空闲毫秒数和起始 ID。之后应使用返回游标继续扫描，直到返回 0-0。某一批认领结果为空，并不代表扫描已经完成；过一段时间再次扫描，原来尚未达到空闲阈值的条目也可能变得可认领。 ...

背压把消费者的有限容量与生产者行为联系起来。通道已满时，生产者可能等待、拒绝新任务，或执行经过明确选择的丢弃策略。关键是容量限制真正改变上游怎么做。队列不断增长，或仪表盘显示积压，只能说明供需失衡，本身不自动形成控制接入的机制。需要从观察到过载，继续追到生产者是否确实减速或停止提交。 坏掉的样子 无界队列可以吸收短暂突发，却也会隐藏持续过载。如果到达速度长期超过完成速度，未完成任务和等待时间就持续增长。把积压挪到待处理线程、重试任务或另一个无界缓冲，并没有消除压力。容量边界必须覆盖任务真正被保存的位置，不能只限制一个容易观察的队列对象，却允许周围继续无限保留工作。 有界队列本身也没有定义完整过载策略。无限阻塞可能耗尽生产者线程或让关闭流程卡住；立即拒绝时，调用方若误认为成功，就可能丢失工作；每次拒绝后立即重试，又可能加重负载。应依据应用契约明确选择：在截止时间内等待、拒绝并稍后重试，还是有意丢弃。不同选择改变交付语义，不能在实现中含糊处理。 队列容量与处理完成是两件事。消费者取出项目可以释放槽位，但此时可能仍在处理这个项目。等待容量为二，也可能同时还有额外执行中的工作。生产者成功入队，仅说明队列接受了项目，不证明已经执行、持久保存或完成下游副作用。若把所有这些状态都叫作“成功”，过载和失败时就无法判断责任归属。 快速验证 将脚本保存为 check-backpressure.py，运行 python3 check-backpressure.py。它只使用 Python 标准库，在 Python 3.14.4 上执行。线程通过明确事件和有界等待协调，不依赖远端消息代理、数据库或刻意拖慢的外部服务。所有队列和任务都只存在当前进程中，结果用于观察接入与完成的语义，不代表分布式队列的持久化保证： from queue import Queue,Full from threading import Event,Thread import json unbounded=Queue() for item in range(1000):unbounded.put_nowait(item) assert unbounded.qsize()==1000 channel=Queue(maxsize=2) channel.put('a');channel.put('b') try: channel.put('deadline-rejected',timeout=.02) raise AssertionError('a full bounded queue accepted another item') except Full:rejected=True assert channel.qsize()==2 entered,delivered=Event(),Event() def produce(): entered.set();channel.put('c',timeout=2);delivered.set() producer=Thread(target=produce,daemon=True);producer.start();assert entered.wait(1) assert not delivered.wait(.05) and channel.qsize()==2 first=channel.get(timeout=1);assert first=='a' assert delivered.wait(1);producer.join(timeout=1);assert not producer.is_alive() assert channel.qsize()==2 rest=[channel.get(timeout=1),channel.get(timeout=1)];assert rest==['b','c'] assert channel.empty() completed=Event() def join():channel.join();completed.set() waiter=Thread(target=join,daemon=True);waiter.start();assert not completed.wait(.05) for item in [first,*rest]:channel.task_done() assert completed.wait(1);waiter.join(timeout=1);assert not waiter.is_alive() print(json.dumps({'unbounded_backlog':unbounded.qsize(),'bounded_capacity':2,'deadline_rejected':rejected, 'blocked_producer_resumed_after_get':delivered.is_set(),'delivered_items':[first,*rest], 'empty_queue_did_not_mean_processing_complete':True,'join_finished_after_task_done':completed.is_set()})) 第一个队列使用默认无界容量。没有消费者运行时，一千次 put_nowait 全部成功，留下同样数量的积压。这是小规模行为演示，不是耗尽内存测试。它说明接入成功本身无法证明下游已有进展，也不能证明到达速度可持续。监控若只统计提交成功率，就可能在处理能力不足时仍显示表面正常。 ...

Redis 列表是有顺序的字节值序列，可以从两端插入和取出。选择哪一端写、哪一端读，决定了它表现为队列还是栈。阻塞操作只是等待数据到达，并不自动提供分布式锁、消息确认或完整的任务处理协议。这些能力必须根据业务需要另外明确，不能从“支持阻塞”直接推导出来。 先把两端的约定说清楚 RPUSH 配合 LPOP 按先进先出顺序移除元素，早进入的元素先离开；RPUSH 配合 RPOP 则是后进先出。一条 LPUSH 带多个值时，会依照参数顺序逐个往左插入，因此传入 a、b、c 后，从左到右看到的是 c、b、a。设计队列时要验证生产者真正执行的命令，而不是只看图里画了哪一端。 列表允许重复值，相同消息插入两次就占两个位置，不会自动完成去重。当两次投递必须能够区分时，应使用不同的任务标识。LRANGE 的起止位置都包含在返回范围里，-1 表示最后一个元素。最后一个元素被移除时，整个列表键也会消失；对不存在的列表弹出会返回空结果，而不是创建一个空列表占位。 两端操作的单元素成本较低，但返回很大的范围或一次删除许多元素，仍然需要相应的工作量。LTRIM 可以保留最近一段窗口，不过用它裁剪待处理任务，就等于主动丢弃被裁掉的条目。生产者写入和随后的单独裁剪是两条操作，不能当成原子提交。保留哪一段应由数据用途决定，不能把日志窗口策略直接套到任务队列上。 等待的是连接，不是整个服务器 BLPOP 在数据到来或等待超时前阻塞当前连接，其他连接仍可正常执行，生产者也能照常投递。在 redis-py 中，成功结果包含键名和取出的值，超时返回 None；超时参数为零则表示无限等待。客户端的套接字超时需要容纳预期的服务器等待时间及响应时间，否则可能先在客户端超时。 消费者没有业务工作时，阻塞调用仍会占用连接。因此应给消费者设置有界的连接预算，并避免让生产者与正在等待的消费者共用同一条连接。还要区分移除顺序与完成顺序：FIFO 保证先移走较早的元素，多个工作者却可能先完成后来的任务。仅凭阻塞等待，无法保证某个业务资源只有一个持有者。 坏掉的样子 LPOP 和 BLPOP 都在业务处理完成之前把任务从列表里移除。消费者刚取到任务就崩溃，或者服务器弹出后响应丢失，该任务都已经不在待处理列表中。Redis 不会自动把它恢复回来，也不要求消费者确认完成。单独配置持久化，并不能修补这类应用层交接窗口。 Redis 6.2 开始提供 LMOVE，可以在一条原子命令中把任务从待处理列表转移到处理中列表；BLMOVE 是对应的阻塞版本。成功处理后，可通过 LREM 按唯一任务标识移除一个条目，作为应用约定的确认。对于长时间没有完成的任务，应用仍需定义恢复策略，因为列表不会自动为每个元素维护租约、时间戳或重试次数。 如果业务副作用已经完成，但确认操作失败，恢复时就可能重复执行。因此唯一任务标识、幂等处理和迟到工作者的处置策略都很重要，两个列表之间的移动不等于恰好执行一次。下面只把一个已知任务明确放回待处理列表，用来验证机制，不把手动操作冒充自动故障检测或生产级恢复调度器。 ...

队列是生产方与消费方之间关于「投递」的一份承诺。这份承诺可以是「至少一次」(永不丢失),也可以是「至多一次」(永不重复),但绝不会两者都免费。选哪个保证,就是选你能容忍哪种失败。 丢失与重复之间的取舍 队列的本质是一份「关于消息会变成什么结果」的承诺，而它决定了你必须接受什么失败。至多一次（at-most-once）承诺丢掉也不重发：赢了就是发一次，输了这个消息跟你再无关系；至少一次（at-least-once）保证消息不丢，但重试会带来重复；至于恰好一次（exactly-once）要在通道外部配上一个幂等消费者或事务标记才可能真正成立。大多数时候你并不会直选，而是由 broker、消费方和处理完的确认方式一起共同决定。 真正的分野藏在确认（ack）这个动作里。消费方先读后确认，这一小段窗口里进程一旦挂了，broker 就会把消息再投一次，把「至少一次」转化为重复；而消费方在处理失败时悄悄跳过，队列就退化成「至多一次」。判断一份设计里的保证，不是读它文档上写什么，而是老老实实看消费方在哪一格 ack。 幂等让「至少一次」安全 「至少一次」听起来吓人，实际上用得极多，因为只要消费方幂等，重复就不是事故。让重复无害的，是一个能把同一份工作做两次却得到同一结果的处理函数：交易用事务号去重、插入天然已被唯一约束挡住。把幂等性直接编码进消息处理，等于把「至少一次」偶发造成的重复消解成一件无关痛痒的小事。 幂等不是一颗塞进去就生效的魔法糖，它是函数层的刻意设计。经典做法是把「已处理」标记一起落库：重试时先查到这条消息已做过，直接返回，而不是再写一遍。对于有昂贵副作用的工作，要把去重标记和副作用放同一个事务里提交，并且确保最终结果由「重复投递的那一次」或「预期的那一次」一致地决定。真实故障下，偶尔重发一次的代价，永远比丢一条消息的代价便宜。 按任务能容忍什么来选 选哪个队列承诺，取决于这个任务即便出错也容得了什么，而不是哪个标签听起来最可靠。指令类的计数、支付这类功能，要求消息绝不丢，应当用「至少一次 + 幂等消费」；埋点这种丢一次高频样本也无关痛痒的，则可以用「至多一次」来省下无界重复带来的存储膨胀。绝大多数系统声称的恰好一次其实都不真实，与其说它是现成的，不如说它是「至少一次 + 幂等 + 持久化投递表」拼出来的一种行为。 要把决定落到代码的行动上，而不是口头：丢消息和重复哪边更伤、消费方能不能廉价地做幂等、broker 在重启后到底保留什么——三条答案定了，就把选择写进文档，再加一个会重放旧消息、并验证消费方吃得下的测试。承诺是对整个系统的契约，而不是贴在面板上的标签。 坏掉的样子 最贵的故障是重复扣款：同一笔订单被收两次，而 broker 监控一切正常。根因是处理完消息、还没发出确认时进程崩溃，broker 把已做完的工作又投一次。区分方法是拿账本对消息号：两笔收费带着同一个号就是重投，带着不同号就是生产方发重了，这是至少一次队列最典型的账单事故。 第二种是消费停滞：吞吐掉到零，某一个位点被无限重试。这是必然抛错的毒消息，配上无限重试、又没有死信队列。与之不同的是重投风暴：吞吐依然很高，但同一批消息反复出现，因为确认丢失，或可见超时短于处理耗时。看重投计数就能区分：只有一条消息计数极高是毒消息，多条消息计数一起上涨是确认丢失，前者要隔离毒消息，后者要修确认链路。 更慢的一种是去重过期：安静几个月后重复突然又生效，因为幂等窗口已滑出，或去重表被清空截断。账本上会看到很久以前的老号被第二次收费，这时不要当成新缺陷去查业务逻辑，先按老号反查账本，去重窗口和清理任务才是源头。 快速验证 把下面代码存为 delivery_guarantee_probe.py，运行 python3 delivery_guarantee_probe.py。 def main(): inbox = [("pay-1", 100), ("pay-2", 50)] deliveries = inbox + [inbox[0]] # lost ack redelivers pay-1 once naive = list(deliveries) # non-idempotent: bills every delivery seen = set() ledger = [] for msg_id, amount in deliveries: if msg_id not in seen: seen.add(msg_id) ledger.append((msg_id, amount)) assert len(naive) == 3, naive # duplicate was billed twice assert sum(a for _, a in ledger) == 150, ledger # idempotent bills pay-1 once assert seen == {"pay-1", "pay-2"}, seen # dedup holds one entry per id print("PASS: naive=3 idempotent=2 dedup=2") if __name__ == "__main__": main() 三个断言分别证明：非幂等把重投算了两次，幂等只算一次，去重集合恰好两个号。预期输出 PASS: naive=3 idempotent=2 dedup=2。 ...