我现在正在使用Python实现一个数据订阅器,它订阅了一个数据发布器(实际上是一个ZeroMQ发布器套接字),并在有新消息时得到通知。在我的订阅者中,收到消息后会将其转储到数据处理器中。当处理完成后,订阅者也将得到处理器的通知。由于数据处理器是用C++编写的,所以我必须使用简单的C++模块扩展Python代码。
以下是我的数据订阅者的简化可运行代码示例。代码中,模块proc表示处理器,订阅了上的ZeroMQ套接字,设置回调,通过调用将接收到的消息发送到处理器。
以下是我的数据订阅者的简化可运行代码示例。代码中,模块proc表示处理器,订阅了上的ZeroMQ套接字,设置回调,通过调用将接收到的消息发送到处理器。
#!/bin/python
# main.py
import gevent
import logging
import zmq.green as zmq
import pub
import proc
logging.basicConfig( format='[%(levelname)s] %(message)s', level=logging.DEBUG )
SUB_ADDR = 'tcp://localhost:10000'
def setupMqAndReceive():
'''Setup the message queue and receive messages.
'''
ctx = zmq.Context()
sock = ctx.socket( zmq.SUB )
# add topics
sock.setsockopt_string( zmq.SUBSCRIBE, 'Hello' )
sock.connect( SUB_ADDR )
while True:
msg = sock.recv().decode( 'utf-8' )
proc.onMsg( msg )
def callback( a, b ):
print( '[callback]', a, b )
def main():
'''Entrance of the module.
'''
pub.start()
proc.setCallback( callback )
'''A simple on-liner
gevent.spawn( setupMqAndReceive ).join()
works. However, the received messages will not be
processed by the processor.
'''
gevent.spawn( setupMqAndReceive )
proc.start()
模块 proc
通过三个导出函数进行了简化:
setCallback
设置回调函数,以便在处理消息时通知我的订阅者;onMsg
被订阅者调用;start
设置一个新的工作线程处理来自订阅者的消息,并使主线程等待工作线程退出。
完整的源代码可以在 GitHub 找到:https://github.com/more-more-tea/python_gil。然而,它并没有按照我的预期运行。一旦处理器线程被添加,订阅者就无法在 gevent 循环中从发布者接收数据。如果我简单地删除数据处理模块,则订阅者的 gevent 循环可以接收来自发布者的消息。
代码有问题吗?我怀疑 GIL 干扰了消息处理中 pthread 的并发性,或者 gevent 循环饿死了。关于这个问题或如何调试它的任何提示都将不胜感激!
setupMqAndReceive
调度为一个线程.Thread而不是gevent.Greenlet,在pthread_join之前释放GIL,并在执行Python回调时在procThread中确保GIL。 - Summer_More_More_Tea