Python异步函数调用方法？异步调用最简实现及任务队列需求

问题1：如何在Python中异步调用函数？ #

在Python里实现异步调用函数，最常用的工具是标准库中的asyncio——它专门用来处理异步IO场景（比如网络请求、文件读写这类耗时但不占CPU的操作）。如果你的action是CPU密集型任务（比如大量计算），那可以结合concurrent.futures的线程池/进程池来异步执行，避免阻塞主线程。

核心思路是：把耗时任务从主线程剥离，让主线程可以继续处理用户输入等操作，同时后台异步执行任务。

问题2：异步调用函数的最简实现方式（满足随时输入+任务入队） #

结合你给出的代码示例，咱们需要实现：用户可以随时输入新值，每个action调用必须进入队列等待处理。这里用asyncio.Queue（异步队列）+ 线程池（处理同步耗时任务）的组合是最简且实用的方案，代码如下：

代码实现（适配Python3） #

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 模拟你的耗时action函数
def action(i):
    print(f"开始处理任务: {i}")
    # 这里替换成实际的耗时操作（比如网络请求、文件处理）
    import time
    time.sleep(3)
    print(f"完成任务: {i}")

# 异步工作线程：持续从队列取任务执行
async def worker(task_queue, executor):
    while True:
        task_data = await task_queue.get()
        # 用线程池执行同步的action，避免阻塞异步事件循环
        await asyncio.get_event_loop().run_in_executor(executor, action, task_data)
        task_queue.task_done()

# 异步输入处理：接收用户输入并加入队列
async def input_handler(task_queue):
    while True:
        # 用run_in_executor包装input，避免阻塞事件循环
        user_input = await asyncio.get_event_loop().run_in_executor(None, input, "Input your value: ")
        await task_queue.put(user_input)
        print(f"任务 {user_input} 已加入队列")

async def main():
    # 创建异步队列，管理所有待处理任务
    task_queue = asyncio.Queue()
    # 创建线程池，可根据需求调整最大并发数
    executor = ThreadPoolExecutor(max_workers=3)
    
    # 启动工作线程（可以启动多个，提升并发处理能力）
    asyncio.create_task(worker(task_queue, executor))
    
    # 启动输入处理，持续接收用户输入
    await input_handler(task_queue)

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

代码说明 #

• 异步队列：asyncio.Queue保证了每个用户输入的任务都会被加入队列，不会丢失，且任务按顺序（或并发）被处理。
• 线程池：因为你的action是同步耗时函数，用线程池可以让它在后台线程执行，不会阻塞异步事件循环，这样用户可以随时输入新值。
• 输入处理：用run_in_executor包装input函数，避免同步输入阻塞整个异步流程。

如果action可以改成异步函数（更简洁） #

如果你的action可以改造成异步函数（比如用asyncio.sleep代替time.sleep），那可以去掉线程池，代码更简洁：

import asyncio

async def action(i):
    print(f"开始处理任务: {i}")
    await asyncio.sleep(3)  # 替换成实际的异步耗时操作
    print(f"完成任务: {i}")

async def worker(task_queue):
    while True:
        task_data = await task_queue.get()
        await action(task_data)
        task_queue.task_done()

async def input_handler(task_queue):
    while True:
        user_input = await asyncio.get_event_loop().run_in_executor(None, input, "Input your value: ")
        await task_queue.put(user_input)
        print(f"任务 {user_input} 已加入队列")

async def main():
    task_queue = asyncio.Queue()
    asyncio.create_task(worker(task_queue))
    await input_handler(task_queue)

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

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