同步异步队列

Question

我计划有一个基于异步队列的生产者-消费者实现，用于处理实时数据，其中以正确的时间顺序发送数据至关重要。下面是它的代码片段：

async def produce(Q, n_jobs):
    for i in range(n_jobs):
        
        print(f"Producing :{i}")
        await Q.put(i)


async def consume(Q):
    while True:
        n = await Q.get()
        
        print(f"Consumed :{n}")
       
       x = do_sometask_and_return_the_result(n)
       print(f"Finished :{n} and Result: {x}")


async def main(loop):
    Q = asyncio.Queue(loop=loop, maxsize=3)
    await asyncio.wait([produce(Q, 10), consume(Q), consume(Q), consume(Q)])
    print("Done")

在这里，生产者生成数据并将其放入异步队列。我有多个消费者来消费和处理数据。在查看输出的同时，在打印"Consumed :{n}“时保持顺序(如1,2,3,4...依此类推)，这完全没问题。但是，由于函数do_sometask_and_return_the_result(n)需要可变的时间来返回结果，因此在下一次打印n "Finished :{n}“时不会保持顺序(如2,1,4,3,5，...)。

有没有办法同步这些数据，因为我需要保持打印结果的顺序？我想看到1,2,3,4，..连续打印'n‘，即使在do_sometask_and_return_the_result(n)之后也是如此。

Answer 1

您可以使用优先级队列系统(使用python heapq库)在作业完成后对其进行重新排序。就像这样。

# add these variables at class/global scope
priority_queue = []
current_job_id = 1
job_id_dict = {}

async def produce(Q, n_jobs):
    # same as above

async def consume(Q):
    while True:
        n = await Q.get()
        
        print(f"Consumed :{n}")
       
       x = do_sometask_and_return_the_result(n)
       await process_result(n, x)


async def process_result(n, x):
    heappush(priority_queue, n)
    job_id_dict[n] = x
    while current_job_id == priority_queue[0]:
        job_id = heappop(priority_queue)
        print(f"Finished :{job_id} and Result: {job_id_dict[job_id]}")
        current_job_id += 1
     


async def main(loop):
    Q = asyncio.Queue(loop=loop, maxsize=3)
    await asyncio.wait([produce(Q, 10), consume(Q), consume(Q), consume(Q)])
    print("Done")

有关heapq模块的详细信息，请访问：https://docs.python.org/3/library/heapq.html