Python 3：如何向线程池提交异步函数？

Question

我想同时使用ThreadPoolExecutorfromconcurrent.futures和 async 函数。

我的程序反复向线程池提交具有不同输入值的函数。在那个更大的函数中执行的最终任务序列可以是任何顺序，我不关心返回值，只关心它们在未来的某个时刻执行。

所以我尝试这样做

async def startLoop():

    while 1:
        for item in clients:
            arrayOfFutures.append(await config.threadPool.submit(threadWork, obj))

        wait(arrayOfFutures, timeout=None, return_when=ALL_COMPLETED)

提交的函数是：

async def threadWork(obj):
   bool = do_something() # needs to execute before next functions
   if bool:
       do_a() # can be executed at any time
       do_b() # ^

wheredo_b和do_ais async 函数。问题是我收到错误：TypeError: object Future can't be used in 'await' expression如果我删除等待，我会收到另一个错误，提示我需要添加await.

我想我可以让一切都使用线程，但我真的不想那样做。

Answer 1

我建议仔细阅读 Python 3 的asyncio 开发指南，特别是“并发和多线程”部分。

您示例中的主要概念问题是事件循环是单线程的，因此在线程池中执行异步协程是没有意义的。事件循环和线程交互的方式有以下几种：

• 每个线程的事件循环。例如：

   async def threadWorkAsync(obj):
       b = do_something()
       if b:
           # Run a and b as concurrent tasks
           task_a = asyncio.create_task(do_a())
           task_b = asyncio.create_task(do_b())
           await task_a
           await task_b
  
   def threadWork(obj):
       # Create run loop for this thread and block until completion
       asyncio.run(threadWorkAsync())
  
   def startLoop():
       while 1:
           arrayOfFutures = []
           for item in clients:
               arrayOfFutures.append(config.threadPool.submit(threadWork, item))
  
           wait(arrayOfFutures, timeout=None, return_when=ALL_COMPLETED)
  

• 在执行程序中执行阻塞代码。这允许您使用异步期货而不是上述并发期货。

   async def startLoop():
       while 1:
           arrayOfFutures = []
           for item in clients:
               arrayOfFutures.append(asyncio.run_in_executor(
                   config.threadPool, threadWork, item))
  
           await asyncio.gather(*arrayOfFutures)
  

• 使用线程安全函数将任务提交到跨线程的事件循环。例如，您可以在主线程的运行循环中运行所有异步协程，而不是为每个线程创建一个运行循环：

   def threadWork(obj, loop):
       b = do_something()
       if b:
           future_a = asyncio.run_coroutine_threadsafe(do_a(), loop)
           future_b = asyncio.run_coroutine_threadsafe(do_b(), loop)
           concurrent.futures.wait([future_a, future_b])
  
   async def startLoop():
       loop = asyncio.get_running_loop()
       while 1:
           arrayOfFutures = []
           for item in clients:
               arrayOfFutures.append(asyncio.run_in_executor(
                   config.threadPool, threadWork, item, loop))
  
           await asyncio.gather(*arrayOfFutures)
  

  注意：这个例子不应该按字面意思使用，因为它会导致所有协程在主线程中执行，而线程池工作线程只是阻塞。这只是为了展示该run_coroutine_threadsafe()方法的一个例子。