Python ZeroMQ 中的异步 Client/Server 模式

Question

我有3个Python写的程序需要连接。 2 个程序 X 和 Y 收集一些信息，这些信息由它们发送给程序 Z。程序 Z 分析数据并将一些决策发送给程序 X 和 Y。类似X和Y的节目数量将在未来扩展。最初我使用命名管道允许从 X、Y 到 Z 的通信。但是正如您所看到的，我需要双向关系。我的老板告诉我使用 ZeroMQ。我刚刚为我的用例找到了模式，称为异步 Client/Server。请参阅下面的 ZMQ 书籍 (http://zguide.zeromq.org/py:all) 中的代码。

问题是我的老板不想使用任何线程、分支等。我将客户端和服务器任务移到单独的程序中，但我不确定如何处理 ServerWorker class。这可以在没有线程的情况下以某种方式使用吗？另外，我想知道如何确定最佳工人数量。

import zmq
import sys
import threading
import time
from random import randint, random

__author__ = "Felipe Cruz <felipecruz@loogica.net>"
__license__ = "MIT/X11"

def tprint(msg):
    """like print, but won't get newlines confused with multiple threads"""
    sys.stdout.write(msg + '\n')
    sys.stdout.flush()

class ClientTask(threading.Thread):
    """ClientTask"""
    def __init__(self, id):
        self.id = id
        threading.Thread.__init__ (self)

    def run(self):
        context = zmq.Context()
        socket = context.socket(zmq.DEALER)
        identity = u'worker-%d' % self.id
        socket.identity = identity.encode('ascii')
        socket.connect('tcp://localhost:5570')
        print('Client %s started' % (identity))
        poll = zmq.Poller()
        poll.register(socket, zmq.POLLIN)
        reqs = 0
        while True:
            reqs = reqs + 1
            print('Req #%d sent..' % (reqs))
            socket.send_string(u'request #%d' % (reqs))
            for i in range(5):
                sockets = dict(poll.poll(1000))
                if socket in sockets:
                    msg = socket.recv()
                    tprint('Client %s received: %s' % (identity, msg))

        socket.close()
        context.term()

class ServerTask(threading.Thread):
    """ServerTask"""
    def __init__(self):
        threading.Thread.__init__ (self)

    def run(self):
        context = zmq.Context()
        frontend = context.socket(zmq.ROUTER)
        frontend.bind('tcp://*:5570')

        backend = context.socket(zmq.DEALER)
        backend.bind('inproc://backend')

        workers = []
        for i in range(5):
            worker = ServerWorker(context)
            worker.start()
            workers.append(worker)

        poll = zmq.Poller()
        poll.register(frontend, zmq.POLLIN)
        poll.register(backend,  zmq.POLLIN)

        while True:
            sockets = dict(poll.poll())
            if frontend in sockets:
                ident, msg = frontend.recv_multipart()
                tprint('Server received %s id %s' % (msg, ident))
                backend.send_multipart([ident, msg])
            if backend in sockets:
                ident, msg = backend.recv_multipart()
                tprint('Sending to frontend %s id %s' % (msg, ident))
                frontend.send_multipart([ident, msg])

        frontend.close()
        backend.close()
        context.term()

class ServerWorker(threading.Thread):
    """ServerWorker"""
    def __init__(self, context):
        threading.Thread.__init__ (self)
        self.context = context

    def run(self):
        worker = self.context.socket(zmq.DEALER)
        worker.connect('inproc://backend')
        tprint('Worker started')
        while True:
            ident, msg = worker.recv_multipart()
            tprint('Worker received %s from %s' % (msg, ident))
            replies = randint(0,4)
            for i in range(replies):
                time.sleep(1. / (randint(1,10)))
                worker.send_multipart([ident, msg])

        worker.close()

def main():
    """main function"""
    server = ServerTask()
    server.start()
    for i in range(3):
        client = ClientTask(i)
        client.start()

    server.join()

if __name__ == "__main__":
    main()

Answer 1

所以，您从这里获取了代码：Asynchronous Client/Server Pattern

请密切注意显示此代码所针对模型的图像。特别是，请查看 "Figure 38 - Detail of Asynchronous Server"。 ServerWorker class 正在旋转 5 "Worker" 个节点。在代码中，这些节点是线程，但您可以使它们完全独立于程序。在那种情况下，您的服务器程序（可能）不会负责启动它们，它们会单独启动并仅与您的服务器通信它们已准备好接收工作。

您会在 ZMQ 示例中经常看到这一点，这是一个在单个可执行文件中模拟线程的多节点拓扑。这只是为了让阅读整个事情变得容易，并不总是打算那样使用。

对于您的特定情况，让工作线程成为线程或将它们分解成单独的程序可能是有意义的……但如果这是您老板的业务要求，那么只需将它们分解成单独的程序。

当然，要回答你的第二个问题，如果不了解他们将要执行的工作量以及他们需要多快做出响应，就无法知道有多少员工是最佳的...你的目标是让工人比收到新工作更快地完成工作。在许多情况下，这很有可能只需要一个工人就可以完成。如果是这样，您可以让您的服务器本身成为工作人员，并跳过整个 "worker tier" 架构。为了简单起见，您应该从这里开始，然后进行一些负载测试，看看它是否真的能有效地应对您的工作负载。如果没有，了解完成一项任务需要多长时间，以及任务进来的速度有多快。假设一名工人可以在 15 秒内完成一项任务。那是每分钟 4 个任务。如果任务以每分钟 5 个任务的形式出现，则您需要 2 个工人，并且您将有一点增长空间。如果情况变化很大，那么您将不得不就资源与可靠性做出决定。

在你走得更远之前，请务必阅读第 4 章，可靠的 Request/Reply 模式，它将提供一些处理异常的见解，并可能为你提供更好的模式。

Python ZeroMQ 中的异步 Client/Server 模式

Asynchronous Client/Server pattern in Python ZeroMQ

python

zeromq