任务队列与背压

更新 2026-10-05

我的学习 →

正在读取学习记录…

任务队列把接收请求与执行工作拆开。生成报告的请求先获得任务标识进入队列，后台工作者逐个领取；用户可以查询排队、处理中和完成状态。队列吸收短时间的流量波动，但工作持续到达得比处理快时，等待时间仍会增长。

背压是处理能力不足时向上游传递压力，例如拒绝新的任务、降低提交速率或让请求稍后再试。容量限制必须明确算哪些任务，否则“工作者领取了任务”可能只是把内存压力从等待队列移到处理中集合。

把容量固定为两个任务

程序依次提交A、B、C三个报告任务。A被工作者领取后，再次提交C；接着模拟A失败，重新入队，由另一位工作者领取并确认完成，随后提交C和重复A。

本例容量统计等待与处理中任务的总数。队列用 deque（双端队列）保存等待顺序，它支持从队尾加入任务、从队首快速领取；inflight（处理中）保存任务与工作者的归属，done 保存已完成任务标识。

容量包含等待与处理中任务，领取不释放容量；失败重新投递还需要业务幂等 — 图1：领取不释放容量；失败重新投递还需要业务幂等。

保存完整程序并运行

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-queue" -Force
cd "F:\学习\l4-queue"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建 F:\学习\l4-queue 文件夹，把下面完整代码保存为 queue_demo.py，然后在PowerShell运行：

cd "F:\学习\l4-queue"
ls
python -X utf8 queue_demo.py

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-queue" -Force
cd "F:\学习\l4-queue"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""用有容量上限的队列模拟报告任务的接收与处理。"""
from collections import deque


class TaskQueue:
    def __init__(self, capacity=2):
        if type(capacity) is not int or capacity < 1:
            raise ValueError("容量应为正整数")
        self.capacity = capacity
        self.ready = deque()
        self.inflight = {}
        self.done = set()

    def submit(self, task_id):
        # 在排队、处理中或已完成的同一任务都不再次入队。
        if task_id in self.ready or task_id in self.inflight or task_id in self.done:
            return "duplicate"
        if len(self.ready) + len(self.inflight) >= self.capacity:
            return "busy"
        self.ready.append(task_id)
        return "queued"

    def claim(self, worker):
        if not self.ready:
            return None
        task_id = self.ready.popleft()
        self.inflight[task_id] = worker
        return task_id

    def ack(self, task_id, worker):
        if self.inflight.get(task_id) != worker:
            raise ValueError("工作者不拥有该任务")
        del self.inflight[task_id]
        self.done.add(task_id)

    def requeue(self, task_id, worker):
        if self.inflight.get(task_id) != worker:
            raise ValueError("工作者不拥有该任务")
        del self.inflight[task_id]
        self.ready.appendleft(task_id)


def run_demo():
    queue = TaskQueue()
    lines = [f"提交A: {queue.submit('A')}", f"提交B: {queue.submit('B')}",
             f"提交C: {queue.submit('C')}"]
    task = queue.claim("worker-1")
    lines.append(f"领取: {task}, 再提交C: {queue.submit('C')}")
    # 失败任务重新入队；业务动作需要另外提供幂等保护。
    queue.requeue(task, "worker-1")
    task = queue.claim("worker-2")
    queue.ack(task, "worker-2")
    lines.append(f"重领并确认: {task}, 重新提交C: {queue.submit('C')}")
    lines.append(f"再次提交A: {queue.submit('A')}, 等待队列={list(queue.ready)}")
    return lines


if __name__ == "__main__":
    print("\n".join(run_demo()))

完整固定输出如下：

提交A: queued
提交B: queued
提交C: busy
领取: A, 再提交C: busy
重领并确认: A, 重新提交C: queued
再次提交A: duplicate, 等待队列=['B', 'C']

为什么领取A后，C仍然被拒绝

第三行返回 busy，因为A和B已占用两份容量。第四行A虽然已离开等待队列，仍在处理中，所以C继续被拒绝。第五行A确认完成后释放容量，C才获得 queued。最后一行重复提交A得到 duplicate，等待队列按顺序保留B和C。

claim 只改变任务归属，ack 才确认本次处理完成。确认需要核对工作者身份；其他工作者无法确认自己未领取的任务。程序在业务完成后调用确认，避免先确认、后失败导致任务丢失。

失败为什么会再次投递

requeue 把已领取任务放回队首，演示一次失败后的再次领取。真实消息队列常以可见性超时或租约处理工作者崩溃：超过时间未确认，就允许其他工作者继续。此时原工作者可能只是变慢，两个执行者都有机会产生副作用，因此业务入口必须识别重复任务。

这里的 done 防止正常完成后再次提交，但不能替代外部业务幂等。工作者写入报告后、确认前退出，队列仍会重新投递；报告写入应使用任务标识作唯一键，必要时结合版本或栅栏编号拒绝旧执行者。

背压反馈怎样到达用户

返回 busy 时还没有入队，调用方应知道任务未被接收，可以稍后用同一任务标识再提交。已经接收的任务则返回当前状态，不能要求用户每次刷新都生成新任务。前端可以显示等待数量和预计耗时；预计值需要根据真实服务吞吐计算。

容量2是演示值。真实容量应按可接受等待时间、任务平均耗时和工作者数量确定，并记录拒绝数、队列深度、最老任务等待时间。只监控队列长度会遗漏单个超长任务。无界队列看似减少拒绝，却可能让所有用户等待到超时。

重投递、失败隔离与持久化

不断失败的任务需要尝试上限，达到上限后转入失败队列供排查，避免长期占据队首。大型任务可以采用公平调度或按业务分队列；优先级也要防止低优先级任务一直等不到执行。

实验用内存集合，重启会失去等待、归属和完成记录，没有实现租约计时、并发锁和磁盘持久化。测试覆盖容量统计、三个阶段的去重、归属检查和重新投递。阅读Temporal 任务执行实现时，可沿任务领取、超时和确认寻找这些边界。

改一处输入，核对机制变化

将 run_demo 中 TaskQueue() 改为 TaskQueue(3)，运行 python -X utf8 queue_demo.py。第三行C变为queued；领取A后再次提交C变为duplicate，因为C已在队列中。恢复默认容量2并重跑，第三行回到busy。容量决定接收量，任务编号决定去重。

运行配套测试

将下面测试保存为同一文件夹的 test_queue_demo.py。它使用固定场景检查状态与业务结果，可以直接运行：

python -X utf8 -m unittest -v test_queue_demo.py

完整测试：test_queue_demo.py

import unittest
from queue_demo import TaskQueue


class QueueTests(unittest.TestCase):
    def test_capacity_includes_claimed_work(self):
        # 领取后仍占容量，只有确认完成才允许新任务进入。
        queue = TaskQueue(1)
        self.assertEqual(queue.submit("a"), "queued")
        queue.claim("w")
        self.assertEqual(queue.submit("b"), "busy")
        queue.ack("a", "w")
        self.assertEqual(queue.submit("b"), "queued")

    def test_duplicate_through_lifecycle(self):
        queue = TaskQueue()
        queue.submit("a")
        self.assertEqual(queue.submit("a"), "duplicate")
        queue.claim("w")
        self.assertEqual(queue.submit("a"), "duplicate")
        queue.ack("a", "w")
        self.assertEqual(queue.submit("a"), "duplicate")

    def test_wrong_worker_cannot_ack(self):
        queue = TaskQueue()
        queue.submit("a")
        queue.claim("w")
        with self.assertRaises(ValueError):
            queue.ack("a", "other")
        self.assertIn("a", queue.inflight)

    def test_redelivery(self):
        queue = TaskQueue()
        queue.submit("a")
        queue.claim("w")
        queue.requeue("a", "w")
        self.assertEqual(queue.claim("other"), "a")
        queue.ack("a", "other")
        self.assertIsNone(queue.claim("other"))


if __name__ == "__main__":
    unittest.main()

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“任务队列与背压”，哪种做法更稳妥？

对应知识点：L4-queue

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L4-queue

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L4-queue

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L4-queue

你的回答