基于队列的通信：发送返回队列是个好主意吗？

Question

我正在使用以下方法来等待在另一个线程中运行的异步操作的结果：

def client(worker_queue, message):
    answer_queue = queue.Queue(maxsize=1)
    worker_queue.put((answer_queue, message))
    result = answer_queue.get(timeout=10)


def worker():
    while True:
        answer_queue, message = worker_queue.get()
        result = do_someting_with(message)
        answer_queue.put(result)
        worker_queue.task_done()

（primitiv worker 只是一个示例。在其他情况下，可能需要在多个回调之间传递“answer_queue”）

这是个好主意，还是我会遇到问题（例如内存管理）？

有没有更好的方法来做到这一点？

我知道asyncio 有类似futures 之类的东西来处理此类问题，但目前我正在寻找（也）适用于多线程的东西。

Answer 1

Python 使用内存地址来访问类变量及其方法。使用队列只是确保每个工作人员都有一个唯一的内存地址位置来放置其答案。

确保，如果变量仍然是局部变量，当您不再需要该变量时，您重新分配它，以便垃圾收集可以释放队列正在使用的内存，或者使用 del 关键字手动清除内存比如del answer_queue。

你可以使用任何数据类型来完成在worker和client之间传输数据的作用，因为所有的类方法都是通过内存地址访问的；然而，最常见的数据传输方式是：

使用队列

queue.Queue() 已经非常优化并且非常通用，因此它是一种在客户端和工作人员之间执行通信的可靠方式。

import queue
import datetime
from threading import Thread
def queue_client(worker_queue, message):
    answer_queue = queue.Queue(maxsize=1)
    worker_queue.put((answer_queue, message))
    result = answer_queue.get(timeout=60)
    print('Bytes received from worker:', len(result))

def queue_worker(worker_queue):
    answer_queue, message = worker_queue.get()
    do_something = lambda x: x[::-1]
    result = do_something(message)
    answer_queue.put(result)

if __name__ == '__main__':
    large_message = 'X' * (2<<30) #2GB

    worker_queue = queue.Queue()
    client = Thread(target=queue_client, args=(worker_queue, large_message,))
    worker = Thread(target=queue_worker, args=(worker_queue,))
    start_time = datetime.datetime.now()
    client.start()
    worker.start()
    client.join()
    worker.join()
    dt = datetime.datetime.now() - start_time
    print('Time elapsed using Queue:', dt.total_seconds()) #~1.47 secs on my machine

共享内存类型

由于在多线程时可以直接访问变量地址，因此可以使用共享内存；这通常是访问数据的最快方式，但您需要某种容器对象来存储数据。

一个非常基本的示例（工人生成、答案提交等可以添加到管理器）：

import datetime
from threading import Thread

class DataManager:

    def __init__(self,):
        self.tasks = {
            #worker_id: message
        }
        self.answers = {
            #worker_id: answer
        }

def shared_client(datamanager, worker_id, message):
    datamanager.tasks[worker_id] = message
    while not datamanager.answers.get(worker_id, None): #Wait for answer
        pass
    result = datamanager.answers[worker_id]
    print('Bytes received from worker:', len(result))
    

def shared_worker(data_manager, worker_id):
    while not data_manager.tasks.get(worker_id, None): #Wait for task to get assigned
        pass

    message = data_manager.tasks[worker_id]
    do_something = lambda x: x[::-1]
    result = do_something(message)
    data_manager.answers[worker_id] = result

if __name__ == '__main__':
    large_message = 'X' * (2<<30) #2GB
    data_manager = DataManager()

    worker_id = 0
    client = Thread(
        target=shared_client, 
        args=(data_manager, worker_id, large_message,)
    )
    worker = Thread(target=shared_worker, args=(data_manager, worker_id))
    start_time = datetime.datetime.now()
    client.start()
    worker.start()
    client.join()
    worker.join()
    dt = datetime.datetime.now() - start_time
    print('Time elapsed using Shared Memory:', dt.total_seconds()) #~1.44 secs on my machine

如果您使用共享内存类型/ctypes，例如 Multiprocessing.Value 和 Multiprocessing.Array (link)，则可能会有更优雅（甚至可能更快）的解决方案。

管道

管道充当连接，一端监听而另一端发送数据。 重要的是，当发送大于 ~32MB 的数据包时，这比队列更有利，而 multiprocessing。此外，它是可序列化的，不像queue.Queue()。
使用管道的示例：

import datetime
from threading import Thread
from multiprocessing import Pipe

def client(conn, message):
    conn.send(message)      #Send message to worker
    result = conn.recv()    #Receive result
    print('Bytes received from worker:', len(result))

def worker(conn):
    message = conn.recv() #Get message from client
    do_something = lambda x: x[::-1]
    result = do_something(message)
    conn.send(result) #Send result to client

if __name__ == '__main__':
    large_message = 'X' * (2<<30) #2GB
    client_conn, worker_conn = Pipe(duplex=True) #Bidirectional pipe
    client_process = Thread(target=client, args=(client_conn, large_message))
    worker_process = Thread(target=worker, args=(worker_conn,))
    
    start_time = datetime.datetime.now()
    client_process.start()
    worker_process.start()

    client_process.join()
    worker_process.join()
    dt = datetime.datetime.now() - start_time

    print('Time elapsed using Pipe:', dt.total_seconds()) #~9.07 secs on my machine

代理

代理在客户端上注册函数，并允许通过调用底层公开函数的工作人员执行它们。当您的工作人员与您的客户端在不同的计算机上时，这会比较棘手，但对于集群计算来说是必要的。