如何在Trio/Anyio中暂停任务并更新Cookie后恢复执行？

完全可以实现这类需求，核心思路是通过共享同步原语控制任务的暂停/恢复，同时保证Cookie更新操作仅执行一次。以下是针对Trio和Anyio的具体实现方案：

Trio 实现方案 #

利用Trio内置的Lock（保证更新操作互斥）和Event（控制任务启停），可以轻松实现全局任务暂停与Cookie更新的逻辑：

import trio

# 全局状态管理
current_cookies = None
cookie_is_valid = False
# 确保Cookie更新操作仅执行一次
update_lock = trio.Lock()
# 控制任务运行状态：初始为允许运行（事件已设置）
resume_event = trio.Event()
resume_event.set()

async def get_cookies():
    # 模拟实际获取/刷新Cookie的逻辑
    await trio.sleep(1)
    return {"session_id": f"valid_session_{trio.current_time():.0f}"}

async def crawler_task(task_id):
    global current_cookies, cookie_is_valid
    while True:
        # 等待恢复信号：若处于暂停状态，此处会阻塞
        await resume_event.wait()
        
        # 检查Cookie有效性，无效则触发更新流程
        if not cookie_is_valid:
            async with update_lock:
                # 双重检查，避免重复更新
                if not cookie_is_valid:
                    # 暂停所有任务：清除事件，后续任务会阻塞在wait()
                    resume_event.clear()
                    print(f"任务{task_id}发现Cookie过期，启动更新流程")
                    current_cookies = await get_cookies()
                    cookie_is_valid = True
                    # 恢复所有任务：设置事件，所有阻塞任务继续执行
                    resume_event.set()
                    print(f"任务{task_id}完成Cookie更新，恢复全局任务")
        
        # 模拟爬虫业务逻辑
        print(f"任务{task_id}使用Cookie {current_cookies}执行爬取")
        await trio.sleep(2)
        
        # 模拟Cookie过期触发（实际场景可替换为请求返回的过期判断）
        if task_id == 1 and trio.current_time() % 10 < 2:
            cookie_is_valid = False
            print(f"任务{task_id}检测到Cookie过期")

async def main():
    # 初始化Cookie
    global current_cookies, cookie_is_valid
    current_cookies = await get_cookies()
    cookie_is_valid = True
    print("初始Cookie已成功获取")
    
    # 启动多爬虫任务
    async with trio.Nursery() as nursery:
        for i in range(3):
            nursery.start_soon(crawler_task, i+1)

if __name__ == "__main__":
    trio.run(main)

Anyio 实现方案 #

Anyio兼容Trio和asyncio后端，只需将同步原语替换为Anyio的API，逻辑与Trio完全一致：

import anyio

# 全局状态管理
current_cookies = None
cookie_is_valid = False
update_lock = anyio.Lock()
resume_event = anyio.Event()
resume_event.set()

async def get_cookies():
    await anyio.sleep(1)
    return {"session_id": f"valid_session_{anyio.current_time():.0f}"}

async def crawler_task(task_id):
    global current_cookies, cookie_is_valid
    while True:
        await resume_event.wait()
        
        if not cookie_is_valid:
            async with update_lock:
                if not cookie_is_valid:
                    resume_event.clear()
                    print(f"任务{task_id}发现Cookie过期，启动更新流程")
                    current_cookies = await get_cookies()
                    cookie_is_valid = True
                    resume_event.set()
                    print(f"任务{task_id}完成Cookie更新，恢复全局任务")
        
        print(f"任务{task_id}使用Cookie {current_cookies}执行爬取")
        await anyio.sleep(2)
        
        # 模拟Cookie过期触发
        if task_id == 2 and anyio.current_time() % 10 < 2:
            cookie_is_valid = False
            print(f"任务{task_id}检测到Cookie过期")

async def main():
    global current_cookies, cookie_is_valid
    current_cookies = await get_cookies()
    cookie_is_valid = True
    print("初始Cookie已成功获取")
    
    async with anyio.create_task_group() as tg:
        for i in range(3):
            tg.start_soon(crawler_task, i+1)

if __name__ == "__main__":
    anyio.run(main)

关键实现要点 #

• 互斥锁(Lock)：确保Cookie更新操作仅由一个任务执行，避免重复请求更新
• 事件(Event)：实现全局任务的暂停/恢复，任务每次执行前会等待事件触发
• 双重检查有效性：在获取锁后再次确认Cookie状态，防止等待锁期间其他任务已完成更新

内容的提问来源于stack exchange，提问作者αԋɱҽԃ αмєяιcαη