python - Python 客户端调用中的双向 ZeroRPC 导致 AssertionError

Question

我的设置有一个 Node.js 子 Node ，它创建了一个带有 Python 对象的 2-way ZeroRPC session 对。

python端类似这样:

class MyClass:
    def __init__(self, socketpath):
        self.client = zerorpc.Client()
        self.client.connect(socketpath)

    def sendtoclient(self, msg):
        self.client.receiveMessage(msg)

if __name__ == '__main__':
    zpc = zerorpc.Server(MyClass(sys.argv[1]))
    zpc.bind(sys.argv[1] + "_python")
    zpc.run()

Node.js 子客户端可以调用 Python 服务器上的方法，但该服务器内的客户端无法在不出现异常的情况下调用 Node.js 子服务器:

Traceback (most recent call last):
File "/usr/lib64/python2.6/site-packages/gevent/queue.py", line 271, in _unlock getter.switch(getter)

File "/usr/lib64/python2.6/site-packages/gevent/hub.py", line 534, in switch assert getcurrent() is self.hub, "Can only use Waiter.switch method from the Hub greenlet"

AssertionError: Can only use Waiter.switch method from the Hub greenlet
<callback at 0x3055e90 args=()> failed with AssertionError

是否需要将 Python 类中的客户端生成为 gevent，然后在需要时调用其 receiveMessage 方法？还是我忽略了其他一些技巧？

Answer 1

经过一些试验，Gevent 文档中的一些示例提供了一个可行的解决方案。我采用的解决方案是创建一个 gevent 队列，该队列从服务器端填充并从循环中的单独 Greenlet 输出。所以在我的服务器文件中，我添加了:

import gevent
from gevent.queue import Queue, Empty

messagesOut = Queue()
def clientWorker(address):
    client = zerorpc.Client()
    if (None != client):
        client.connect(address)
        while True:
            try:
                messages = messagesOut.get()
                client.passMessages(messages) # a method on the Node.js-side ZeroRPC server
            except: # Empty could be thrown here if you're interested
                pass
            finally:
                gevent.sleep(0)

在 MyClass 中，它的初始化更改为保留对队列的引用作为 self.outbox(实际上，我可以在每次访问它时都使用 global)。当需要发送异步消息时，MyClass 调用 self.outbox.put(messages)。

然后在下面创建 ZeroRPC 实例时，我生成了每个实例:

if __name__ == '__main__':
    ge_client = gevent.spawn(clientWorker, sys.argv[1] + "_node2python")
    zpc = zerorpc.Server(messagesOut)
    zpc.bind(sys.argv[1] + "_python2node")
    ge_server = gevent.spawn(zpc.run)
    gevent.joinall([ge_client, ge_server]) # up and running here.

作为概念验证，执行上述操作解决了我的问题。