如何用Asyncio运行两类异步函数：一类返回数据，一类无返回值

直接上可运行的代码示例，核心是用asyncio.create_task托管状态检查的后台任务，同时用asyncio.gather批量发起API请求，两者完全并行：

import asyncio
import random
from datetime import datetime

class SessionManager:
    def __init__(self):
        self.session_status = "active"  # 要更新的类属性

    async def check_session_status(self, check_interval=2):
        """按设定频率检查并更新会话状态"""
        while True:
            # 模拟状态检查逻辑，实际可替换为调用状态API/读取缓存
            self.session_status = random.choice(["active", "warning", "active"])
            print(f"[{datetime.now()}] 会话状态更新为: {self.session_status}")
            await asyncio.sleep(check_interval)

async def make_request(api_url="https://example.com/api"):
    """模拟异步API调用，返回响应对象"""
    # 模拟请求耗时
    await asyncio.sleep(random.uniform(1, 3))
    # 模拟响应数据，实际可替换为aiohttp等库的真实请求逻辑
    return {
        "url": api_url,
        "status_code": 200,
        "response_data": f"Response from {api_url} at {datetime.now()}"
    }

async def main():
    # 初始化会话管理器
    session_manager = SessionManager()
    
    # 启动状态检查后台任务
    status_task = asyncio.create_task(session_manager.check_session_status())
    
    try:
        # 并行发起4次API调用，等待所有请求完成
        responses = await asyncio.gather(
            *[make_request() for _ in range(4)]
        )
        
        # 处理所有API响应
        print("\n=== 所有API响应处理结果 ===")
        for idx, resp in enumerate(responses, 1):
            print(f"请求 {idx}: {resp['response_data']}，当前会话状态: {session_manager.session_status}")
    finally:
        # 任务完成后取消状态检查任务
        status_task.cancel()
        try:
            await status_task
        except asyncio.CancelledError:
            print("\n状态检查任务已终止")

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

关键逻辑说明 #

• 后台状态检查：asyncio.create_task会把状态检查函数放入事件循环后台运行，不阻塞API请求执行，通过await asyncio.sleep控制检查频率
• 并行API调用：asyncio.gather接收多个协程对象，同时启动所有API请求，直到全部完成后返回响应列表
• 状态共享：asyncio是单线程事件循环，直接修改类属性session_status是安全的，API请求处理时可直接读取最新状态
• 任务清理：用finally块确保状态检查任务在API请求完成后被取消，避免程序退出时残留后台任务

扩展提示 #

• 真实API调用可替换make_request内的模拟逻辑为aiohttp.ClientSession的请求代码
• 可给状态检查任务添加异常捕获，避免单次检查出错导致任务终止
• 若API调用需使用会话状态（如携带token），直接在make_request中读取session_manager的对应属性即可

内容的提问来源于stack exchange，提问作者Coldchain9