Python run_in_executor 忘了？

Question

我怎样才能设置一个阻塞函数在执行器中运行，结果并不重要，所以主线程不应该等待或被它拖慢。

说实话，我不确定这是否是正确的解决方案，我想要的只是将某种类型的处理队列与主进程分开，这样它就不会阻止服务器应用程序返回请求，因为这种类型的 Web 服务器为多个请求运行一个工作器。

我最好远离像 Celery 这样的解决方案，但如果这是最理想的，我愿意学习它。

这里的上下文是一个异步网络服务器，它生成带有大图像的 pdf 文件。

app = Sanic()
#App "global" worker
executor = ProcessPoolExecutor(max_workers=5)

app.route('/')
async def getPdf(request):
  asyncio.create_task(renderPdfsInExecutor(request.json))
  #This should be returned "instantly" regardless of pdf generation time
  return response.text('Pdf being generated, it will be sent to your email when done')

async def renderPdfsInExecutor(json):
  asyncio.get_running_loop.run_in_executor(executor, syncRenderPdfs, json)

def syncRenderPdfs(json)
  #Some PDF Library that downloads images synchronously
  pdfs = somePdfLibrary.generatePdfsFromJson(json)
  sendToDefaultMail(pdfs)

上面的代码给出了错误（是的，它以管理员身份运行）：

PermissionError [WinError 5] Access denied
Future exception was never retrieved

额外的问题：通过在执行程序中运行异步循环，我能获得什么吗？因此，如果它一次处理多个 PDF 请求，它将在它们之间分配处理。如果是，我该怎么做？

Answer 1

好的，所以首先有一个误解。这个

async def getPdf(request):
    asyncio.create_task(renderPdfsInExecutor(request.json))
    ...

async def renderPdfsInExecutor(json):
    asyncio.get_running_loop.run_in_executor(executor, syncRenderPdfs, json)

是多余的。做就够了

async def getPdf(request):
    asyncio.get_running_loop.run_in_executor(executor, syncRenderPdfs, request.json)
    ...

或者（因为你不想等待）更好

async def getPdf(request):
    executor.submit(syncRenderPdfs, request.json)
    ...

现在你遇到的问题是因为syncRenderPdfs 抛出了PermissionError。它没有被处理，所以 Python 会警告你“嘿，一些后台代码抛出了一个错误。但是代码不属于任何人，所以这到底是怎么回事？”。这就是你得到Future exception was never retrieved 的原因。 您的 pdf 库本身有问题，而不是 asyncio。一旦你解决了这个内部问题，确保安全也是一个好主意：

def syncRenderPdfs(json)
    try:
        #Some PDF Library that downloads images synchronously
        pdfs = somePdfLibrary.generatePdfsFromJson(json)
        sendToDefaultMail(pdfs)
    except Exception:
        logger.exception('Something went wrong')  # or whatever

您的“权限被拒绝”问题完全不同，您应该对其进行调试和/或为此发布一个单独的问题。

至于最后一个问题：是的，executor 会排队并在worker之间平均分配任务。

编辑：正如我们在 cmets 中谈到的那样，实际问题可能出在您使用的 Windows 环境中。或者更准确地说，使用 ProcessPoolExecutor，即生成进程可能会更改权限。我建议使用 ThreadPoolExecutor，假设它在平台上运行良好。

Answer 2

您可以查看 asyncio.gather(*tasks) 以并行运行多个。

请记住，并行任务只有在 io 绑定且不阻塞时才能正常工作。

python 文档中的一个示例 (https://docs.python.org/3/library/asyncio-task.html)：

import asyncio

async def factorial(name, number):
    f = 1
    for i in range(2, number + 1):
        print(f"Task {name}: Compute factorial({number}), currently i={i}...")
        await asyncio.sleep(1)
        f *= i
    print(f"Task {name}: factorial({number}) = {f}")
    return f

async def main():
    # Schedule three calls *concurrently*:
    L = await asyncio.gather(
        factorial("A", 2),
        factorial("B", 3),
        factorial("C", 4),
    )
    print(L)

asyncio.run(main())