Python Asyncio：确保后台任务顺序执行的策略

import asyncio
import random

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

async def collect_data():
    """模拟数据收集操作，耗时随机。"""
    event_loop = asyncio.get_event_loop()
    while True:
        print("我正在收集数据...")
        await asyncio.sleep(random.randint(1, 5)) # 模拟收集耗时
        # 直接创建后台任务保存数据，可能导致多个保存任务同时运行
        event_loop.create_task(save_data())

# 运行主程序
# asyncio.run(collect_data()) # 如果运行此代码，会发现"我正在保存一个批次的数据..."可能重复出现，且完成顺序不确定

import asyncio
import random

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

async def collect_data_sequential_save():
    """数据收集，确保保存任务顺序执行。"""
    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 = event_loop.create_task(save_data_batch())

        # 为了演示，可以设置一个退出条件
        # if random.random() < 0.1:
        #     break

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

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

import asyncio
import random

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

async def save_all_batches(queue: asyncio.Queue):
    """消费者：从队列中取出数据并保存。"""
    while True:
        try:
            # 从队列中获取一个批次，如果队列为空则等待
            batch = await queue.get() 
            print(f"消费者：从队列中取出批次 {batch}，准备保存。")
            await save_data_batch() # 执行保存操作
            queue.task_done() # 标记此任务已完成
        except asyncio.CancelledError:
            # 当消费者任务被取消时，优雅退出
            print("消费者任务被取消，退出。")
            break
        except Exception as e:
            print(f"消费者：保存过程中发生错误: {e}")
            queue.task_done() # 即使出错也要标记完成，避免死锁

async def collect_data_with_queue():
    """生产者：收集数据并放入队列。"""
    event_loop = asyncio.get_event_loop()
    # 创建一个有最大容量的队列，防止内存溢出
    queue = asyncio.Queue(maxsize=4) 

    # 启动消费者任务
    saving_task = event_loop.create_task(save_all_batches(queue))

    batch_id = 0
    try:
        while True:
            print(f"生产者：我正在收集数据 (批次 {batch_id})...")
            await asyncio.sleep(random.randint(1, 3)) # 模拟收集耗时

            # 将收集到的数据（这里用批次ID代替）放入队列
            # 如果队列已满，此put操作会等待直到有空间
            await queue.put(f"Batch-{batch_id}") 
            print(f"生产者：批次 {batch_id} 已放入队列。队列当前大小: {queue.qsize()}")
            batch_id += 1

            # 为了演示，在收集一定数量批次后停止
            if batch_id >= 10:
                print("生产者：已收集足够批次，停止收集。")
                break
    finally:
        # 收集完成后，等待队列中的所有任务完成
        print("生产者：等待所有队列中的批次保存完成...")
        await queue.join() 
        print("生产者：所有队列中的批次已保存完成。")

        # 取消消费者任务，使其优雅退出
        saving_task.cancel()
        # 确保消费者任务被取消并完成清理
        await saving_task

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

    # 生产者在put时监控消费者状态的更健壮示例片段
    # (这只是概念性代码，实际可能更复杂)
    # putting_task = event_loop.create_task(queue.put(f"Batch-{batch_id}"))
    # done, pending = await asyncio.wait({putting_task, saving_task},
    #                                  return_when=asyncio.FIRST_COMPLETED)
    # if saving_task in done:
    #     # 消费者任务已完成（可能因错误），取消put操作并退出
    #     putting_task.cancel()
    #     print("消费者任务已意外退出，停止生产。")
    #     break
    # else:
    #     # put操作完成，继续生产
    #     await putting_task # 确保put操作的异常被捕获