如何在asyncio python中使用subprocess模块限制并发进程数量

Question

import asyncio
import asyncio.subprocess
args="blah blah argument "     
create=asyncio.create_subprocess_shell(args,stdout=asyncio.subprocess.PIPE)
proc = await create
output= await proc.stdout.read( )

这是我的服务器代码的一部分，它会从客户端接收数千个并行请求。现在我该如何限制服务器创建的子进程的最大数量，以运行参数 blah blah。由于此代码正在使用我的 CPU 的 100％，我需要在同一 CPU 上部署其他服务器。

- Aravind

---

考虑使用类似Celery和RabbitMQ这样的工具值得吗？从长远来看，它可能会为您提供更多的可扩展性和更精细的控制。只是一个想法。 - Andrew Guy

客户端通过websocket连接。而且我运行的进程是CPU绑定的，所以我需要异步支持，这使我更喜欢asyncio。Celery和RMQ是否支持这一点？我从未使用过它们 :/ - Aravind

Celery是一个异步任务队列，而RabbitMQ是任务代理。本文提供了一个不错的例子 - http://suzannewang.com/celery-rabbitmq-tutorial/。 - Andrew Guy

如果你之前没有接触过，这可能不适合你的应用程序，但值得探索。 - Andrew Guy

Answer 1

asyncio.Semaphore 是一种限制同时进行作业数量的内部计数器的方法：

sem = asyncio.Semaphore(10)

async def do_job(args):
    async with sem:  # Don't run more than 10 simultaneous jobs below
        proc = await asyncio.create_subprocess_shell(args, stdout=PIPE)
        output = await proc.stdout.read()
        return output

注意，您应该确保工作数量的增长速度不要比您实际能够完成它们的速度快得多。否则，您将需要更复杂的解决方案。

- Mikhail Gerasimov