Как запустить корутину в бесконечном цикле в ProcessPoolExecutor

Question

У меня есть корутина, в которой я подписываюсь на события в шине NATS:

import asyncio
import os

from nats.aio.client import Client as NATS
from nats.aio.errors import ErrNoServers

async def run(async_loop):
    nc = NATS()

    try:
        await nc.connect(servers=['nats://localhost:4222'], loop=async_loop)
    except ErrNoServers as e:
        print(e)
        return

    async def msg_handler(msg):
        subject = msg.subject
        reply = msg.reply
        data = msg.data.decode()
        print(f"[PID: {os.getpid()}]  Received a message on '{subject} {reply}': {data}")

    await nc.subscribe("foo.>", cb=msg_handler, queue='awesome.queue')

Я могу сделать из нё задачу и запустить в бесконечном цикле:

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.create_task(run(loop))
    loop.run_forever()

Однако, мне хочется запустить несколько процессов параллельно для обработки схожих эвентов. Есть ли возможность используя ProcessPoolExecutor запустить в бесконечных циклах мои воркеры?

Answer 1

def process():
    loop = asyncio.get_event_loop()
    loop.create_task(run(loop))
    loop.run_forever()

if __name__ == '__main__':
    with concurrent.futures.ProcessPoolExecutor() as executor:
         executor.submit(process)
         executor.submit(process)
         executor.submit(process)

БЛОГ НА HUSL

Как запустить корутину в бесконечном цикле в ProcessPoolExecutor

Ответы (1 шт):