在 Python RQ

编程问答 2022-05-18

问题描述

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

我重新编写了我的用户作业以使用 aiohttp 而不是 python 请求，而且速度明显更快。在 Python RQ 文档中，它说“任何 Python 函数调用都可以放在 RQ 队列中”。但我不知道如何将我的异步函数发送到队列中？


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，它只是一个协程期货......我如何在队列中等待我的函数？

解决方法

由于 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、线程或类似可以与同步函数一起使用的东西。

async-awaitpythonpython-rq