Python：manager.Queue() 与 asyncio。如何解决死锁？

Question

我试图弄清楚如何让基于 websocket 的服务器监听传入的请求，将它们放入队列中以供另一个进程工作，然后将结果放入另一个队列中，基于 websocket 的服务器可以等待所述结果并将响应发送回客户端。

这只是我试图学习并获得更多关于异步和进程之间共享数据的经验。我正在使用 Python 3.9.2 64 位。

现在我陷入了死锁，正如服务器代码中的“producer_handler”函数中所评论的那样。这是我正在玩的代码：

服务器：

import asyncio
import logging
import time
from multiprocessing import Manager, Process
import websockets

logging.root.setLevel(0)



def server(recievequeue, sendqueue):


    async def consumer_handler(websocket, path):
        while True:
            logging.info('Waiting for request')
            try:
                request = await websocket.recv()
            except Exception as exception:
                logging.warning(f'consumer_handler Error: {exception}')
                break
            logging.info(f'Request: {request}')
            recievequeue.put(request)
            logging.info('Request placed in recievequeue')


    async def producer_handler(websocket, path):
        while True:
            logging.info('Waiting for response')
            response = sendqueue.get()# Deadlock is here.
            try:
                await websocket.send(response)
            except Exception as exception:
                logging.warning(f'producer_handler Error: {exception}')
                break
            logging.info('Response sent')


    async def handler(websocket, path):
        consumer_task = asyncio.ensure_future(consumer_handler(websocket, path))
        producer_task = asyncio.ensure_future(producer_handler(websocket, path))

        done, pending = await asyncio.wait([producer_task, consumer_task], return_when=asyncio.FIRST_COMPLETED)

        for task in done:
            logging.info(f'Canceling: {task}')
            task.cancel()

        for task in pending:
            logging.info(f'Canceling: {task}')
            task.cancel()


    eventloop = asyncio.get_event_loop()
    eventloop.run_until_complete(websockets.serve(handler, 'localhost', 8081, ssl=None))
    eventloop.run_forever()



def message_handler(recievequeue, sendqueue):
    while True:
        # I just want to test getting a message from the recievequeue, and placing it in the sendqueue
        request = recievequeue.get()
        logging.info(f'Request: {request}')
        time.sleep(3)
        data = str(time.time())
        logging.info(f'Work completed @ {data}')
        sendqueue.put(data)



def main():
    logging.info('Starting Application')

    manager = Manager()
    sendqueue = manager.Queue()
    recievequeue = manager.Queue()

    test_process_1 = Process(target=server, args=(recievequeue, sendqueue), name='Server')
    test_process_1.start()

    test_process_2 = Process(target=message_handler, args=(recievequeue, sendqueue), name='Message Handler')
    test_process_2.start()

    test_process_1.join()



if __name__ == '__main__':
    main()

还有客户：

import asyncio
import logging
import websockets

logging.root.setLevel(0)
URI = "wss://localhost:8081"



async def test():


    async def consumer_handler(connection):
        while True:
            try:
                request = await connection.recv()
            except Exception as exception:
                logging.warning(f'Error: {exception}')
                break
            logging.info(request)


    async def producer_handler(connection):
        while True:
            await asyncio.sleep(5)
            try:
                await connection.send('Hello World')
            except Exception as exception:
                logging.warning(f'Error: {exception}')
                break


    async with websockets.connect(URI, ssl=None) as connection:
        consumer_task = asyncio.ensure_future(consumer_handler(connection))
        producer_task = asyncio.ensure_future(producer_handler(connection))
        while True:
            await asyncio.wait([consumer_task, producer_task], return_when=asyncio.FIRST_COMPLETED)



def main():
    logging.info('Starting Application')

    eventloop = asyncio.get_event_loop()

    try:
        eventloop.run_until_complete(test())
        eventloop.run_forever()
    except Exception as exception:
        logging.warning(f'Error: {exception}')



if __name__ == '__main__':
    main()

如果我删除队列，服务器和多个客户端可以毫无问题地来回交谈。我只是不知道如何获取（）和放置（）请求和响应。任何帮助将不胜感激！

Answer 1

所以在浏览了其他帖子后，我注意到其他人在谈论死锁并使用run_in_executor。经过更多测试后，我发现用以下代码替换导致死锁的行解决了问题：

response = await eventloop.run_in_executor(None, sendqueue.get)