Python异步队列asyncio.Queue在消费者崩溃后的消息丢失

在基于 asyncio.Queue 构建的生产者-消费者系统中，消费者协程意外崩溃是一个常见但棘手的问题。默认情况下，asyncio.Queue 提供的“取出即消费”的语义，意味着当一个任务从队列中被取出（queue.get()）后，如果消费者在处理该任务时崩溃，该任务就会永久丢失。这不仅会导致业务逻辑中断，还可能引发数据不一致。本指南将手把手地为你构建一个具有消息持久化和崩溃恢复能力的健壮异步队列系统。

1. 理解问题：为什么消息会丢失

一个标准的异步队列消费者流程如下：

1. 生产者调用 await queue.put(item) 放入消息。
2. 消费者调用 item = await queue.get() 取出消息。
3. 消费者开始处理 item。
4. 如果在步骤3中，消费者因异常（如 KeyError, TimeoutError 或未捕获的业务异常）而崩溃，那么 item 就从内存中消失了。queue.task_done() 从未被调用，但队列本身也无法知道这个任务“失踪”了。

问题的根源在于内存队列缺乏事务性。取出和完成是两个独立的步骤，没有被捆绑成一个原子操作。

2. 解决方案核心：引入“处理中”状态与确认机制

我们的思路是：

1. 持久化：将消息存储在可重启、可恢复的存储介质中（如数据库或文件），而不仅仅是内存。
2. 状态跟踪：为每条消息引入明确的状态，例如 pending（待处理）、processing（处理中）、completed（已完成）、failed（已失败）。
3. 显式确认：消费者处理成功后，必须显式地将消息状态更新为 completed，这个动作称为“确认”（Ack）。如果消费者崩溃，它未来得及确认，系统可以检测到这些卡在 processing 状态的消息，并将其重新放回队列。

我们将使用一个简单的 sqlite3 数据库作为消息存储，因为它无需额外服务，非常适合演示和中小型应用。

3. 步骤一：构建消息存储与队列包装器

首先，我们创建一个 DurableQueue 类来封装核心逻辑。

import sqlite3
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional

class MessageStatus(Enum):
    PENDING = “pending”
    PROCESSING = “processing”
    COMPLETED = “completed”
    FAILED = “failed”

@dataclass
class Message:
    id: int
    content: Any
    status: MessageStatus

class DurableQueue:
    def __init__(self, db_path: str = “:memory:”, maxsize: int = 0):
        self._db_path = db_path
        self._maxsize = maxsize
        self._conn: Optional[sqlite3.Connection] = None
        self._async_queue = asyncio.Queue(maxsize=maxsize)
        self._setup_db()

    def _setup_db(self):
        """**初始化**数据库连接并**创建**消息表。"""
        self._conn = sqlite3.connect(self._db_path)
        self._conn.execute(“””
            CREATE TABLE IF NOT EXISTS messages (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                content TEXT NOT NULL,
                status TEXT NOT NULL DEFAULT ‘pending’
            )
        “”“)
        self._conn.commit()

    async def put(self, content: Any) -> int:
        """**将**消息**持久化**到数据库，并**放入**内存队列。"""
        cursor = self._conn.execute(
            “INSERT INTO messages (content, status) VALUES (?, ?)”,
            (str(content), MessageStatus.PENDING.value)
        )
        message_id = cursor.lastrowid
        self._conn.commit()
        # 为了复用 asyncio.Queue 的异步接口，我们放入一个代表该消息的对象
        await self._async_queue.put(Message(id=message_id, content=content, status=MessageStatus.PENDING))
        return message_id

    async def get(self) -> Message:
        """**从**内存队列**取出**一个消息，并立即将其状态**更新为** ‘processing’。"""
        message = await self._async_queue.get()
        # 原子操作：取出并标记为处理中
        self._conn.execute(
            “UPDATE messages SET status = ? WHERE id = ?”,
            (MessageStatus.PROCESSING.value, message.id)
        )
        self._conn.commit()
        message.status = MessageStatus.PROCESSING
        return message

    async def ack(self, message: Message):
        """**确认**消息处理成功，**更新**状态为 ‘completed’。"""
        self._conn.execute(
            “UPDATE messages SET status = ? WHERE id = ?”,
            (MessageStatus.COMPLETED.value, message.id)
        )
        self._conn.commit()
        # 通知内存队列该任务已完成
        self._async_queue.task_done()

    async def nack(self, message: Message, requeue: bool = False):
        """**处理**消息失败。如果 requeue=True，则**重新放入**队列，否则标记为 ‘failed’。"""
        if requeue:
            # 重置状态为 pending，以便下次被取出
            self._conn.execute(
                “UPDATE messages SET status = ? WHERE id = ?”,
                (MessageStatus.PENDING.value, message.id)
            )
            self._conn.commit()
            await self._async_queue.put(message) # 重新入队
        else:
            self._conn.execute(
                “UPDATE messages SET status = ? WHERE id = ?”,
                (MessageStatus.FAILED.value, message.id)
            )
            self._conn.commit()
        self._async_queue.task_done()

    async def recover_interrupted_messages(self):
        """**恢复**因消费者崩溃而卡在 ‘processing’ 状态的消息。"""
        cursor = self._conn.execute(
            “SELECT id, content FROM messages WHERE status = ?”,
            (MessageStatus.PROCESSING.value,)
        )
        for row in cursor.fetchall():
            msg_id, content = row
            print(f“[恢复] 发现中断的消息 ID: {msg_id}，将其状态**重置为** pending。”)
            # 将状态改回 pending
            self._conn.execute(
                “UPDATE messages SET status = ? WHERE id = ?”,
                (MessageStatus.PENDING.value, msg_id)
            )
            self._conn.commit()
            # 重新放入内存队列，等待消费者处理
            await self._async_queue.put(Message(id=msg_id, content=content, status=MessageStatus.PENDING))

    def close(self):
        """**关闭**数据库连接。"""
        if self._conn:
            self._conn.close()

4. 步骤二：编写健壮的消费者协程

消费者的核心是必须使用 try...finally 结构，并确保无论处理成功还是失败，都调用 ack() 或 nack()。

async def robust_consumer(queue: DurableQueue, worker_id: int):
    print(f“消费者 {worker_id} **启动**。”)
    while True:
        try:
            # 1. 从持久化队列中安全地获取任务
            message = await queue.get()
            print(f“消费者 {worker_id} **开始处理**消息 {message.id}: {message.content}”)

            # 2. **模拟**业务处理（这里可能崩溃）
            # 例如：处理一个字符串，如果内容包含 “crash” 则抛出异常
            if “crash” in message.content:
                raise ValueError(“模拟业务处理崩溃！”)
            await asyncio.sleep(1) # 模拟耗时操作

            # 3. **处理成功**，显式确认
            await queue.ack(message)
            print(f“消费者 {worker_id} **成功处理**消息 {message.id}。”)

        except asyncio.CancelledError:
            # 协程被取消（例如程序关闭），跳出循环
            print(f“消费者 {worker_id} **被取消**。”)
            break
        except Exception as e:
            # 4. **处理失败**，决定是否重试（此处示例为重试一次）
            print(f“消费者 {worker_id} **处理**消息 {message.id} **失败**，错误: {e}”)
            # 这里可以加入重试计数逻辑，例如 message.retry_count += 1
            # 如果重试次数超过限制，则 requeue=False
            await queue.nack(message, requeue=True) # 暂时设为True，表示重试

5. 步骤三：启动系统与模拟崩溃

编写主程序来启动生产者和消费者，并在中途杀死一个消费者以验证恢复机制。

async def main():
    # **使用**一个文件数据库，以便重启后消息依然存在
    durable_queue = DurableQueue(db_path=“tasks.db”)

    # **恢复**上次遗留的中断任务（重要！）
    await durable_queue.recover_interrupted_messages()

    # **启动**两个消费者
    consumers = [
        asyncio.create_task(robust_consumer(durable_queue, worker_id=1)),
        asyncio.create_task(robust_consumer(durable_queue, worker_id=2))
    ]

    # **生产**一些任务，包括一个会引发崩溃的任务
    tasks_to_produce = [f“task_{i}” for i in range(5)] + [“task_crash_once”]
    for task_content in tasks_to_produce:
        await durable_queue.put(task_content)
        print(f“**已生产**任务: {task_content}”)

    # **等待**一段时间，让消费者处理
    await asyncio.sleep(3)

    # **模拟**消费者1崩溃：我们通过取消其协程来实现
    print(“**模拟**消费者1崩溃……”)
    consumers[0].cancel()
    try:
        await consumers[0]
    except asyncio.CancelledError:
        pass

    # **给**系统一些时间，让恢复机制和剩余消费者工作
    print(“**等待**系统恢复……”)
    await asyncio.sleep(5)

    # **关闭**剩余消费者和队列
    for c in consumers[1:]:
        c.cancel()
    for c in consumers[1:]:
        try:
            await c
        except asyncio.CancelledError:
            pass

    durable_queue.close()
    print(“系统**已关闭**。”)

if __name__ == “__main__”:
    asyncio.run(main())

6. 运行与观察效果

1. 首次运行：你会看到任务被生产，两个消费者交替处理。当处理到 task_crash_once 时，某个消费者会报错，但任务会被 nack 并重新入队（requeue=True）。当消费者1被模拟取消时，它正在处理的任何任务都会因为 CancelledError 而进入 finally 块（如果有的话），但在我们的示例中，CancelledError 会被单独捕获并跳出循环，不会调用 ack 或 nack。这意味着该任务的状态会卡在 processing。
2. 再次运行同一程序：程序启动时会首先调用 recover_interrupted_messages。它会从数据库中找出状态为 processing 的消息（即上次消费者崩溃时遗留的），将其状态重置为 pending，并重新放入内存队列。然后，这些消息会被正常的消费者再次处理。

7. 进阶考虑

• 数据库选择：对于高性能、分布式场景，应将 sqlite3 替换为 Redis（使用列表或流数据结构）、RabbitMQ（使用ACK机制）或 Kafka。核心逻辑（状态跟踪与确认）是相通的。
• 死信队列：对于多次重试失败的消息，应将其路由到“死信队列”进行人工干预，而不是无限重试。
• 并发控制：通过 asyncio.Semaphore 控制同时处理“处理中”消息的数量，避免任务堆积。
• 优雅关闭：在捕获 signal.SIGINT 时，应等待当前正在处理的消息完成或进行超时中断，而不是直接强杀进程。

通过引入持久化存储和显式确认机制，我们成功地将 asyncio.Queue 从一个易失的内存缓冲区，升级为了一个具备基本容错能力的任务队列，有效解决了消费者崩溃导致的消息丢失问题。