在 Python RQ 中执行可等待/异步函数

Question

我的工作都是需要针对每个对象提出的一系列请求。即，它是一个 User，其中包含多个数据点（可能有数百个），需要通过请求将其添加到该用户。我最初将这些请求同步写入运行但它阻塞且缓慢。我将每个 User 作业发送到 Python RQ，并有 10 名工作人员通过用户发送到队列中。 1 个工作人员，1 个用户，阻止请求。

我重写了我的用户作业以使用 aiohttp 而不是 python 请求，而且速度明显更快。在 Python RQ 文档中说 'Any Python function call can be put on an RQ queue.' 但我不知道如何将我的异步函数发送到队列中？


async def get_prices(calls: List[dict]) -> List[dict]:
     async with aiohttp.ClientSession() as session:
         for price in prices.items():
                price_type, date = price
                price = await pg.get_price(
                    session=session, lookup_date=date
                )
        do_some_other_stuff()
        await session.close()


from core.extensions import test_queue
from prices import get_prices
job = test_queue.enqueue(get_prices, kwargs={"username":'username'})

问题是 get_prices 永远不会被等待，它只是一个协程期货....我怎样才能在队列中等待我的函数？

Answer 1

由于 python-rq 不直接支持 asyncio，您可以使用调用 asyncio.run 的同步函数。

async def _get_prices(calls: List[dict]) -> List[dict]:
    # ...

def get_prices(*args, **kwargs):
    asyncio.run(_get_prices(*args, **kwargs))

但是请注意，asyncio.run 仅在没有其他运行事件循环时才有效。如果您希望异步循环已经是运行，请改用 loop.create_task。

def get_prices(*args, **kwargs):
    loop = asyncio.get_event_loop()
    coro = _get_prices(*args, **kwargs)
    loop.create_task(coro)

那么当python-rq调用get_prices时就会导致异步函数被执行。

另一种选择是不使用 asyncio 来发出请求，例如使用 grequests、线程或类似的可以与同步函数一起使用的方法。

在 Python RQ 中执行可等待/异步函数

Executing an awaitable / async function in Python RQ

python

async-await

python-rq