Python Asyncio 中背景任务的顺序执行与并发管理

import asyncio
import random

async def save_data():
    """模拟数据保存操作"""
    print("我正在保存一个批次的数据...")
    await asyncio.sleep(2) # 模拟IO耗时
    print("一个批次的数据保存完成。")

async def collect_data_problematic():
    """存在并发保存问题的示例"""
    event_loop = asyncio.get_event_loop()
    while True:
        print("我正在收集数据...")
        await asyncio.sleep(random.randint(1, 5)) # 模拟数据收集耗时
        # 直接创建任务，可能导致多个save_data并发运行
        event_loop.create_task(save_data())

# asyncio.run(collect_data_problematic())

import asyncio
import random

async def save_data():
    """模拟数据保存操作"""
    print("我正在保存一个批次的数据...")
    await asyncio.sleep(2) # 模拟IO耗时
    print("一个批次的数据保存完成。")

async def collect_data_await_previous():
    """通过等待前一个任务完成来协调"""
    event_loop = asyncio.get_event_loop()
    last_save_task = None # 用于跟踪上一个保存任务

    while True:
        print("我正在收集数据...")
        await asyncio.sleep(random.randint(1, 5)) # 模拟数据收集耗时

        if last_save_task: # 如果存在上一个保存任务
            print("等待上一个保存任务完成...")
            await last_save_task # 阻塞直到上一个保存任务完成
            print("上一个保存任务已完成，可以开始新的保存。")

        # 启动新的保存任务并更新last_save_task
        last_save_task = event_loop.create_task(save_data())

    # 循环结束后，确保最后一个保存任务也完成
    if last_save_task:
        await last_save_task

# 运行示例
# asyncio.run(collect_data_await_previous())

import asyncio
import random

async def save_data_batch(batch_data):
    """模拟数据保存操作，接受批次数据"""
    print(f"我正在保存批次: {batch_data}...")
    await asyncio.sleep(2) # 模拟IO耗时
    print(f"批次: {batch_data} 保存完成。")

async def collect_data_with_queue():
    """使用asyncio.Queue协调数据收集与保存"""
    event_loop = asyncio.get_event_loop()
    # 创建一个有最大容量的队列，防止内存耗尽
    # maxsize=16 表示队列最多能存储16个待处理的批次
    queue = asyncio.Queue(maxsize=16) 

    async def save_all_batches():
        """消费者协程：从队列中取出数据并保存"""
        while True:
            try:
                batch = await queue.get() # 等待并获取一个批次
                await save_data_batch(batch) # 执行保存操作
                queue.task_done() # 标记该任务已完成
            except asyncio.CancelledError:
                # 协程被取消时退出循环，实现优雅关闭
                print("保存任务被取消，退出。")
                break
            except Exception as e:
                print(f"保存任务发生错误: {e}")
                queue.task_done() # 即使出错也要标记完成，防止queue.join()死锁

    # 启动后台保存任务
    saving_task = event_loop.create_task(save_all_batches())

    batch_counter = 0
    while True:
        print("我正在收集数据...")
        await asyncio.sleep(random.randint(1, 5)) # 模拟数据收集耗时
        batch_counter += 1
        batch_data = f"Batch-{batch_counter}" # 模拟收集到的数据批次

        # 将数据放入队列，如果队列满则等待
        print(f"将 {batch_data} 放入队列...")
        await queue.put(batch_data) 
        print(f"{batch_data} 已放入队列。")

    # 优雅关闭：等待所有队列中的任务完成，然后取消保存任务
    # 注意：实际应用中，通常会有一个外部信号来触发退出循环
    # await queue.join() # 等待所有放入队列的任务被处理
    # saving_task.cancel() # 取消后台保存任务
    # await saving_task # 确保取消操作完成