Python多线程ZeroMQ REQ-REP

Question

我希望使用多线程通过 Python 和 ZeroMQ 实现 REQ-REP 模式。

使用 Python，我可以在新客户端连接到服务器时创建一个新线程。该线程将处理与该特定客户端的所有通信，直到套接字关闭:

# Thread that will handle client's requests
class ClientThread(threading.Thread):
    # Implementation...
    def __init__(self, socket):
        threading.Thread.__init__(self)
        self.socket = socket
    def run(self):
        while keep_alive:
            # Thread can receive from client
            data = self.socket.recv(1024)
            # Processing...
            # And send back a reply
            self.socket.send(reply)

while True:
    # The server accepts an incoming connection
    conn, addr = sock.accept()
    # And creates a new thread to handle the client's requests
    newthread = ClientThread(conn)
    # Starting the thread
    newthread.start()

是否可以使用 ZeroMQ 做同样的事情[*]？我看过一些使用 ZeroMQ 和 Python 进行多线程的例子，但是在所有这些例子中，一开始都是用固定数量的线程创建一个线程池，它似乎更倾向于负载平衡。

[*] 请注意，我想要的是保持客户端与其线程之间的连接处于事件状态，因为线程期望来自客户端的多个 REQ 消息，并且它将存储必须在消息之间保留的信息(即:一个变量在新的 REQ 消息上递增其值的计数器；因此每个线程都有自己的变量，任何其他客户端都不应该能够访问该线程)。新客户 = 新话题。

Answer 1

是的，ZeroMQ 是一个功能强大的实用工具箱

然而，最令人惊讶的是，ZeroMQ <socket>-s 比您在示例中使用的普通对应物结构化得多。

{ aZmqContext -> aZmqSocket -> aBehavioralPrimitive }

ZeroMQ 在“单例”ZMQ- Context 的引擎盖下构建了一个非凡的、抽象丰富的框架。 ，这是(并将继续)唯一用作“共享”的东西。

线程不应“共享”任何其他“派生”对象，更不用说它们的状态，因为实现了强大的分布式责任框架架构，无论是为了简洁设计还是高性能和低延迟.

对于所有 ZMQ- Socket -s 人们宁愿想象一个更智能的分层子结构，在那里人们可以减轻对 I/O 事件的担忧(在 ZMQ 中管理- Context 责任 --因此 keep-alive 问题、计时问题和公平队列缓冲/选择轮询问题简单地停止对您来说是可见的......)，具有一种正式的通信模式行为(由选定的 ZMQ- Socket 类型原型(prototype)给出)。

最后

ZeroMQ 和类似的 nanomsg 库是相当像乐高积木的项目，作为架构师和设计师，它赋予您的能力比人们通常在一开始就意识到的要多。

因此，人们可以专注于分布式系统行为，而不是将时间和精力浪费在解决 just-another-socket-messaging-[噩梦] 上。

(绝对值得一读 Pieter Hintjens 的两本书，他是 ZeroMQ 的共同之父。在那里你会发现很多关于这个伟大主题的啊哈!时刻。)

... 作为蛋糕上的樱桃——您将所有这些都视为与传输无关的通用环境，无论是在 inproc:// 上传递一些消息吗？ ，其他超过 ipc:// 并通过 并行听/说 tcp:// 层。

EDIT#1 2014-08-19 17:00 [UTC+0000]

请检查下面的评论，并进一步检查您的 - 初级和高级 - 设计选项，用于 <trivial-failure-prone>-spin-off 处理，<负载平衡>-REP-worker 队列，用于<可扩展>-分布式处理和<故障恢复模式>-REP-worker 二进制文件-开始阴影处理。

没有一堆模型 SLOC，没有一个代码示例可以做到一刀切。

这在设计分布式消息系统时非常有效。

"""REQ/REP modified with QUEUE/ROUTER/DEALER add-on ---------------------------

   Multithreaded Hello World server

   Author: Guillaume Aubert (gaubert) <guillaume(dot)aubert(at)gmail(dot)com>

"""
import time
import threading
import zmq

print "ZeroMQ version sanity-check: ", zmq.__version__

def aWorker_asRoutine( aWorker_URL, aContext = None ):
    """Worker routine"""
    #Context to get inherited or create a new one trick------------------------------
    aContext = aContext or zmq.Context.instance()
    
    # Socket to talk to dispatcher --------------------------------------------------
    socket = aContext.socket( zmq.REP )
    
    socket.connect( aWorker_URL )
    
    while True:

        string  = socket.recv()

        print( "Received request: [ %s ]" % ( string ) )
        
        # do some 'work' -----------------------------------------------------------
        time.sleep(1)

        #send reply back to client, who asked --------------------------------------
        socket.send( b"World" )

def main():
    """Server routine"""

    url_worker = "inproc://workers"
    url_client = "tcp://*:5555"

    # Prepare our context and sockets ------------------------------------------------
    aLocalhostCentralContext = zmq.Context.instance()

    # Socket to talk to clients ------------------------------------------------------
    clients = aLocalhostCentralContext.socket( zmq.ROUTER )
    clients.bind( url_client )

    # Socket to talk to workers ------------------------------------------------------
    workers = aLocalhostCentralContext.socket( zmq.DEALER )
    workers.bind( url_worker )

    # --------------------------------------------------------------------||||||||||||--
    # Launch pool of worker threads --------------< or spin-off by one in OnDemandMODE >
    for i in range(5):
        thread = threading.Thread( target = aWorker_asRoutine, args = ( url_worker, ) )
        thread.start()

    zmq.device( zmq.QUEUE, clients, workers )

    # ----------------------|||||||||||||||------------------------< a fair practice >--
    # We never get here but clean up anyhow
    clients.close()
    workers.close()
    aLocalhostCentralContext.term()

if __name__ == "__main__":
    main()