消息队列削峰完全指南:从一次"秒杀活动瞬间把数据库打挂"看懂异步削峰

2021 年我做一个电商的秒杀功能。需求很简单:某个商品,在活动开始的那一刻,以远低于平时的价格限量开抢。第一版我做得很直白:用户点"立即抢购",后端就同步地把这一整套流程跑完——校验库存、扣减库存、创建订单、写订单表、扣用户余额,全部做完,才给前端返回"抢购成功"。平时测试,一切正常,响应飞快。可活动真正开始的那一秒,系统崩了。我看监控:活动开始前,接口每秒几十个请求,岁月静好;活动开始的那一刻,每秒请求数瞬间冲到了几万。这几万个请求,每一个都触发了那套"校验—扣减—建单—写库"的同步流程,几万个请求同时挤进数据库。数据库的连接池瞬间被占满,后面的请求全部排队、超时,接着是接口大面积 500,然后是整个服务雪崩。我第一反应是"机器不够,加机器;SQL 慢,优化 SQL"。我加了一倍的服务器,把那几条 SQL 翻来覆去地优化,然后等下一次活动——结果还是崩,只是崩得稍微晚了几秒。我盯着那个监控曲线看了很久才彻底想明白:这个问题,加机器和优化 SQL 根本解不了。因为活动开始那一刻的流量,是一根又高又尖的瞬时尖峰——它可能是平时的几百倍,但它只持续几秒钟。我为了扛住这几秒的尖峰,要把整个系统的容量堆到平时的几百倍,那其余 99.99% 的时间,这些机器全在空转,这笔账完全算不过来。而且就算我真堆了那么多机器,数据库那一层的写入能力,也不是靠堆 web 服务器能解决的。我错在一个根本的假设上:我以为"请求来了,就必须立刻把它彻底处理完"。可秒杀这件事,用户点下"抢购"的那一刻,他真正急着要的,只是一个"你的请求我收到了、正在排队"的确认;他的订单到底在哪一毫秒被写进数据库,他并不在意,慢个一两秒他完全能接受。这就是关键的破局点:既然"收到请求"和"处理完请求"这两件事,用户对它们的时间要求天差地别,那我为什么要把它们绑死在一次同步调用里?我可以把它们拆开:请求一来,我飞快地把它记下来(扔进一个队列),立刻就告诉用户"收到了";然后,后端再按照数据库扛得住的节奏,从队列里不慌不忙地一个一个取出来处理。瞬时涌进来的几万个请求,被这个队列接住、缓冲,再被匀速地"喂"给数据库——那根又高又尖的尖峰,就被削平成了一条系统扛得住的平缓曲线。这,就是消息队列削峰。我以为消息队列不过是"扔个东西进去再取出来",结果真做下来坑一个接一个:消息会丢、会重复、会堆积如山……那次之后我才认真把消息队列削峰从头搞明白。这篇文章就把它梳理一遍:为什么同步处理扛不住瞬时峰值、削峰的本质是什么、生产者怎么把请求快速入队、消费者怎么按系统能力匀速消费,以及消息丢失、重复消费、消息积压这些把削峰真正做对要避开的坑。

问题背景

先把那次的现象和我的误判讲清楚,后面所有的设计都是冲着纠正这个误判去的。

现象:一个秒杀接口,平时每秒几十个请求,毫无压力。活动开始的那一刻,每秒请求数瞬间冲到几万。每个请求都同步执行"校验库存—扣减—建单—写库"的完整流程,几万个请求同时挤进数据库,连接池被占满,接口大面积超时、500,服务雪崩。

我当时的错误认知:"扛不住是因为机器不够、SQL 不够快,加机器、优化 SQL 就行了。"

真相:瞬时峰值是一根又高又尖、只持续几秒的尖峰。为扛住它把容量堆到平时的几百倍,绝大部分时间机器都在空转,算不过账,而且数据库写入能力堆 web 服务器也解决不了。真正的破局点是:"收到请求"和"处理完请求"用户对二者的时间要求天差地别,不该把它们绑死在一次同步调用里。用一个消息队列把二者解耦——请求快速入队就返回,后端按数据库扛得住的节奏匀速消费。瞬时尖峰被队列缓冲、被削平成平缓曲线。

要把削峰做好,需要几块认知:

• 为什么同步处理扛不住瞬时峰值,加机器为什么解不了;
• 削峰的本质是怎样用队列把请求和处理解耦;
• 生产者怎么把请求快速写进队列就立刻返回;
• 消费者怎么按系统的能力匀速地把队列里的请求消费掉;
• 消息丢失、重复消费、消息积压这些工程坑怎么处理。

一、为什么同步处理扛不住瞬时峰值

先把这件最根本的事钉死:同步处理,意味着请求的速率,被强行等于了处理的速率。

在同步模型里,一个请求进来,这个请求的线程就一直占着,直到把"校验—扣减—建单—写库"全部做完才释放。也就是说,系统同时能受理多少请求,完全取决于同时能处理完多少请求。下面这段代码,就是我那个同步的秒杀接口:

def seckill_sync(user_id: int, product_id: int):
    # 反面教材:请求一来,就同步地把整套流程全部做完。
    if not check_stock(product_id):            # 1. 校验库存
        return {"ok": False, "msg": "已售罄"}
    deduct_stock(product_id)                   # 2. 扣减库存
    order_id = create_order(user_id, product_id)   # 3. 创建订单
    deduct_balance(user_id)                    # 4. 扣用户余额
    return {"ok": True, "order_id": order_id}
    # 问题:这 4 步全要读写数据库。平时每秒几十个请求没事;
    # 活动开始瞬间几万个请求同时跑这 4 步,几万个请求同时
    # 砸向数据库 —— 连接池占满,全部排队、超时,服务雪崩。

这段代码的错,不在某一行,在它的结构:它让"请求进来的速率",和"数据库处理的速率",变成了同一个数。平时这个数是每秒几十,数据库轻松;活动瞬间它变成每秒几万,而数据库的处理能力没变——它还是只能每秒处理几千。于是每秒几万的请求,挤向每秒只能处理几千的数据库,多出来的那几万,要么排队排到超时,要么直接被拒。

这时候你会本能地想"加机器"。但加机器加的是web 服务器——它能让你"受理"更多请求,可这些请求最终还是要落到同一个数据库上。数据库才是那个真正的瓶颈,而它很难靠简单加机器来等比例放大。更何况,这尖峰只有几秒,为它常备几百倍的机器,完全是浪费。结论很硬:瞬时峰值,不能靠"扩容硬扛"来解决。得换个思路。

二、削峰的本质:用队列把请求和处理解耦

上一节的死结是"请求速率被绑死等于处理速率"。削峰要做的,就是解开这个绑定。

怎么解开?在"请求"和"处理"之间,插进一个队列。请求进来,不再当场处理,而是被飞快地塞进队列,然后就立刻给用户返回——这一步只是一次内存级的写入,极快,所以无论瞬间来多少请求,都能接得住。而真正的处理,由后端的消费者,从队列的另一头,按自己的节奏一个一个取出来做。

这样一来,两个速率就彻底分开了:请求入队的速率,可以瞬间冲到每秒几万——队列扛得住;请求被处理的速率,则稳定在数据库扛得住的每秒几千——消费者匀速地做。中间多出来的请求,就暂存在队列里,排着队,等着被慢慢消费。把这件事画成图就一目了然:入口是一根又高又尖的尖峰,经过队列这个"水库",流出来的是一条平缓的、恒定的水流。水库不能凭空消灭洪水,但它能把洪峰蓄起来,再匀速地放——削峰削的就是这个"峰",它没让请求变少,它让请求不再挤在同一刻。下面几节,就把这个水库的两端造出来:先是入水口(生产者),再是出水口(消费者)。

三、生产者:把请求飞快地写进队列

先造入水口——生产者。它的职责只有一个,而且必须极快:把请求塞进队列。这里用 Redis 的 list 当队列(LPUSH 从一头写入),它是内存操作,足够快:

import redis
import json

r = redis.Redis()
QUEUE_KEY = "seckill:queue"


def enqueue_order(user_id: int, product_id: int) -> bool:
    """生产者:把一次抢购请求塞进队列,这一步极快(内存操作)。"""
    msg = json.dumps({"user_id": user_id, "product_id": product_id})
    r.lpush(QUEUE_KEY, msg)        # 从队列左端推入一条消息
    return True
    # 注意:这里【不碰数据库】,不扣库存、不建订单。
    # 它只做一件事 —— 把请求记下来。所以无论瞬间来多少,都接得住。

有了它,秒杀接口就可以改头换面了。它不再同步跑那四步,而是入队、立刻返回:

def seckill_async(user_id: int, product_id: int):
    """新的秒杀接口:只负责把请求入队,然后立刻返回。"""
    # 可以在这里先做一道极轻量的拦截(比如 Redis 里预扣个数)
    if r.get(f"sold_out:{product_id}"):
        return {"ok": False, "msg": "已售罄"}
    enqueue_order(user_id, product_id)
    # 关键:不等订单真正创建,立刻返回"排队中"
    return {"ok": True, "msg": "抢购请求已提交,正在为你排队处理"}

对比第一节的 seckill_sync:那个接口要等四步数据库操作全做完才返回,慢且重;这个 seckill_async 只做一次 lpush 就返回,快且轻。用户拿到的不再是"抢购成功",而是"正在排队"——这正是我们在第一节想清楚的:用户那一刻急着要的,本就只是一个"收到了"的确认。瞬时几万的请求,到这里就被队列稳稳接住了。接住之后,轮到出水口。

四、消费者:按系统扛得住的节奏匀速处理

现在造出水口——消费者。它在后端独立运行,做一件事:从队列里取出请求,真正地处理。它的核心特征是——匀速。

def consume_loop():
    """消费者:从队列另一头不断取出请求,一个一个真正处理。"""
    while True:
        # BRPOP:从队列右端阻塞式取一条;没有消息就阻塞等待
        item = r.brpop(QUEUE_KEY, timeout=5)
        if item is None:
            continue                      # 5 秒没消息,继续等
        _, raw = item
        msg = json.loads(raw)
        process_order(msg["user_id"], msg["product_id"])

这个循环用了 BRPOP——"阻塞式右端弹出"。它从队列的另一头取消息(生产者从左端进,消费者从右端出,先进先出);队列空时,它阻塞等待,不空转。而它真正处理一条消息的逻辑,就是当初那套同步流程:

def process_order(user_id: int, product_id: int):
    """真正的下单处理:第一节那四步,挪到这里来做。"""
    if not check_stock(product_id):
        r.set(f"sold_out:{product_id}", "1")   # 卖完了,打个标记
        return
    deduct_stock(product_id)                   # 扣减库存
    order_id = create_order(user_id, product_id)   # 创建订单
    deduct_balance(user_id)                    # 扣余额
    notify_user(user_id, f"抢购成功,订单 {order_id}")  # 异步通知用户

关键来了:消费者处理的速率,不由请求的速率决定,而由它自己决定。队列里哪怕积压了十万条消息,消费者也只是一条接一条地取——它取的速度,就是数据库能稳稳扛住的速度。想让处理快一点,就多开几个消费者进程并行消费;但开多少个,要算准:让它们加起来的总消费速率,刚好压在数据库的承受线以下。这就是削峰的精髓:无论入口的洪峰多猛,落到数据库上的,永远是一条由消费者数量精确控制的、平缓的水流。

五、削峰之外:解耦与异步带来的附加好处

队列插进来之后,你会发现它带来的好处不止削峰。

第一个附加好处是解耦。process_order 里有一步 notify_user——抢购成功后通知用户。在同步模型里,如果这个通知服务挂了或者很慢,它会拖垮整个下单接口。但现在,下单和通知被队列隔开了:就算通知环节出问题,也只是通知这一步受影响,核心的下单照常进行。各个环节通过队列连接,一个环节的故障,不会顺着同步调用链扩散。

第二个附加好处是异步带来的体验提升。用户点完抢购,接口瞬间就返回了"排队中",他不用对着转圈的图标干等几秒。下面这个函数,体现的就是消费者可以从容地做一些同步接口里不敢做的、稍慢的事——比如带重试的通知:

import time


def notify_user_with_retry(user_id: int, text: str, max_retry: int = 3):
    """异步环节可以从容地做重试,因为它不占用用户的等待时间。"""
    for attempt in range(max_retry):
        try:
            send_notification(user_id, text)
            return
        except Exception:
            time.sleep(2 ** attempt)      # 失败了,退避一下再重试
    # 重试都失败,记下来,后续人工或定时任务补偿
    log_failed_notification(user_id, text)

这种"失败了退避重试三次"的逻辑,放在同步接口里是不敢想的——它会让用户多等好几秒。但在消费者这一侧,它不占用任何人的等待时间,可以做得很从容。削峰是队列的主业,解耦和异步是它顺带给的红利。主干通了,但要把削峰真正做对,还有几个绕不开的坑。

六、工程坑:消息丢失、重复消费与消息积压

削峰的主干通了,但消息队列有几个非做不可的工程防护。

坑 1:消息不能在处理中途丢失。用 BRPOP 取消息有个隐患:消息一旦被弹出,就从队列里消失了。如果消费者刚弹出消息、还没处理完就崩了,这条消息就彻底丢了——用户的抢购请求凭空蒸发。稳妥的做法是用一个"处理中"的中转队列(Redis 的 BRPOPLPUSH):弹出的同时把消息备份进中转队列,确认处理成功后,再从中转队列删掉。

PROCESSING_KEY = "seckill:processing"


def consume_with_ack():
    """带确认的消费:消息先转入'处理中'队列,处理成功才删除。"""
    # BRPOPLPUSH:从主队列弹出,同时压入"处理中"队列(原子操作)
    raw = r.brpoplpush(QUEUE_KEY, PROCESSING_KEY, timeout=5)
    if raw is None:
        return
    msg = json.loads(raw)
    try:
        process_order(msg["user_id"], msg["product_id"])
        r.lrem(PROCESSING_KEY, 1, raw)    # 处理成功,从"处理中"删除
    except Exception:
        # 处理失败:消息还在"处理中"队列里,不会丢,可被重新捞回
        log_error(f"处理失败,消息保留待重试: {raw}")

坑 2:同一条消息可能被消费两次。消费者处理完、但在删除消息之前崩了,这条消息会被再消费一遍。如果 process_order 不防重,用户就会被扣两次款、下两个单。所以消费逻辑必须幂等——比如给订单加一个由"用户+商品+活动"算出的唯一键,重复处理时,数据库的唯一索引会挡住第二次。这一点本身是个大话题,这里只点明:消费者必须假设每条消息都可能被处理多次。

坑 3:消息会积压,必须监控。如果生产得比消费得快(消费者太少、或处理变慢),队列就会越积越长。积压本身不是错——削峰削的就是靠暂存;但积压如果持续增长不回落,说明消费能力根本跟不上,得赶紧加消费者。所以队列长度必须接入监控:

def check_backlog(warn_threshold: int = 10000):
    """监控队列积压:超过阈值就告警,提示该加消费者了。"""
    backlog = r.llen(QUEUE_KEY)            # 队列里还有多少条没处理
    processing = r.llen(PROCESSING_KEY)    # 还有多少条卡在"处理中"
    if backlog > warn_threshold:
        send_alert(f"队列积压 {backlog} 条,消费能力不足,请加消费者")
    if processing > 0:
        # "处理中"队列里的陈年消息,多半是消费者崩溃留下的,要捞回重试
        send_alert(f"有 {processing} 条消息卡在处理中,需检查并重试")
    return backlog, processing

坑 4:有的消息永远处理不成功,要进死信。某条消息因为数据本身有问题,怎么重试都失败。不能让它一直占着、反复重试拖累整个队列。要给它设一个重试上限,超过了就把它挪进一个专门的"死信队列",由人工去排查,主流程则继续往下走。下面这张图,把一条秒杀请求经过削峰的完整路径串起来:

关键概念速查

避坑清单

1. 同步处理让请求受理速率等于处理速率,瞬时峰值一来几万请求同时砸数据库必然雪崩。
2. 瞬时尖峰只持续几秒,加机器为它常备几百倍容量算不过账,数据库瓶颈也难靠加机器解决。
3. 削峰的本质是在请求和处理之间插队列,把入队速率和处理速率彻底解耦。
4. 生产者只做一件事:把请求飞快写进队列就立刻返回,绝不在入队环节碰数据库。
5. 秒杀接口返回的是"排队中"而非"成功",用户那一刻要的本就只是收到确认。
6. 消费者按数据库扛得住的节奏匀速取出消息,处理速率由消费者数量控制而非请求速率。
7. 队列还顺带解耦:环节间用队列连接,一个环节挂掉不会顺着同步调用链扩散拖垮全局。
8. BRPOP 弹出即消失,消费者中途崩溃消息会丢,要用 BRPOPLPUSH 转入处理中队列再确认删除。
9. 消息可能被消费两次,消费逻辑必须幂等,靠唯一键挡住重复下单重复扣款。
10. 队列积压要接入监控,持续增长就加消费者;屡试屡败的消息设重试上限后移入死信队列。

总结

回头看那次"秒杀活动瞬间把数据库打挂"的事故,以及我后来在消息队列上接连踩的坑,最该记住的不是某一种队列用法,而是我动手前那个想当然的判断——"请求来了,就必须立刻把它彻底处理完"。这句话错在它把"受理一个请求"和"处理完一个请求"当成了同一件事。可它们根本不是:受理,是说一声"我收到了",这件事极快、可以瞬间做几万次;处理,是真刀真枪地读写数据库,这件事很重、每秒只能做几千次。我的第一版把这两件快慢悬殊的事焊死在一次同步调用里,于是慢的那件,就成了快的那件的天花板。削峰做的事,本质上就是把这两件事拆开,中间垫一个队列,让快的归快、慢的归慢。

所以做削峰,真正的工程量不在"塞个东西进队列再取出来"那一下。lpush 一下、brpop 一下,这部分 Demo 里谁都能写。真正的工程量在那个队列的两端和中间:消费者处理到一半崩了,那条消息是稳稳还在,还是凭空蒸发了?同一条消息被取了两次,你的下单逻辑是扛得住,还是给用户扣了两次款?队列越积越长,你是有监控早早告警,还是等用户投诉"抢购了半天没动静"才发现?有那么一条怎么都处理不了的烂消息,它是被你挪进死信队列、还是堵在那里反复重试拖垮所有人?这篇文章的几节,其实就是顺着这条思路展开的:先想清楚同步为什么扛不住峰值,再看削峰的本质是用队列解耦,然后是生产者、消费者这两段主干,最后是消息丢失、重复消费、积压死信这几个把削峰真正做对的工程细节。

你会发现,削峰的思路,和我们生活里应对一切"瞬时拥挤"的经验都是相通的。一个热门景点,门口不会让上万人同时挤进闸机——它放一个蜿蜒的排队通道,人流瞬间涌来,先在通道里排着,再被匀速地放进去。一家爆满的餐厅,不会让所有客人堵在门口,它发取号小票,你拿了号就能去逛街,到号了再回来。那个排队通道、那张取号小票,就是现实世界里的"消息队列"——它们没有让人变少,它们做的是,把"同一刻涌来的人",变成"先后有序到来的人"。你的系统扛不住的从来不是"请求多",而是"请求挤在同一刻";队列要做的,就是把这个"挤",摊平成一个"排"。

最后想说,削峰做没做扎实,差距永远不会在 Demo 里暴露——Demo 里你手动点几下抢购,请求稀稀拉拉地来,有没有队列、有没有那些防护,跑起来一模一样。它只在真实的活动洪峰、真实的消费者宕机、真实的消息积压面前才显形。那时候它会用最难堪的方式给你结账:活动开始的那一秒,你的数据库连接数曲线垂直拔起,接口成片地返回 500,运营在群里发"页面打不开了";一个消费者进程悄无声息地崩了,几百条用户的抢购请求跟着它一起蒸发,事后对账怎么也对不平;一条消息被重复消费,一个用户的余额被扣了两次,然后是投诉、是退款、是排查到深夜。所以别等活动当天系统在你眼前崩掉,在你写下第一个秒杀接口的时候就该想清楚:瞬间几万的请求,我是当场硬接,还是先入队缓冲?消费者崩了,我的消息还在吗?一条消息被取两次,我的下单扛得住吗?队列堆成山,我看得见吗?这几个问题都有了答案,你的系统才不只是 Demo 里那个稀疏请求跑得通的样子,而是一个无论洪峰多高、来得多猛,都能稳稳接住、匀速放行、一单不丢也不重的可靠系统。

—— 别看了 · 2026