python - 在 python 中使用 pyzmq 的日志处理程序

Question

我想在 Python 程序中引入基于 zmq 的日志记录。当我遇到 ZMQError: Address in use 错误时，我决定将其归结为一个简单的概念证明。我能够运行精简版本，但没有收到任何日志条目。这是我使用的代码:

日志发布者:

import time
import logging
from zmq.log import handlers as zmqHandler


logger = logging.getLogger('myapp')
logger.setLevel(logging.ERROR)
zmqH=zmqHandler.PUBHandler('tcp://127.0.0.1:12344')
logger.addHandler(zmqH)
for i in range(50):
    logger.error('error test...')
    print "Send error #%s" % (str(i))
    time.sleep(1)

结果

Send error #0
Send error #1
Send error #2
Send error #3
Send error #4
...

登录订阅者:

import time
import zmq

def sub_client():
    port = "12344"       
    context = zmq.Context()
    socket = context.socket(zmq.SUB)    
    socket.connect("tcp://127.0.0.1:%s" % port)
    # Generate 30 entries
    for i in range (30):
        print "Listening to publishers..."
        message = socket.recv()
        print "Received error #%s: %s" % (str(i), message)
        time.sleep(1)

sub_client()

结果

Listening to publishers...

因此订阅者在调用 socket.recv() 时被锁定。我在不同的控制台中启动了发布者和订阅者。当我使用netstat时，这两个进程都会出现:

C:\>netstat -a -n -o | findstr 12344
  TCP    127.0.0.1:12344        0.0.0.0:0              LISTEN          1336
  TCP    127.0.0.1:12344        127.0.0.1:51937        ESTABLISHED     1336
  TCP    127.0.0.1:51937        127.0.0.1:12344        ESTABLISHED     8624

我在这里看不到我的错误，有什么想法吗？

除了手头的问题之外，我一般如何使用这个 zmq 监听器。我是否必须为每个进程创建一个 PUBHandler 实例，然后将其添加到 logger 的所有实例 (logging.getLogger('myapp') 创建一个自己的记录器实例，对吗？)或者我是否必须为我使用的所有不同类创建一个自己的 PUBHandler ？由于 PUBHandler 类有一个 createLock() 我认为它不是线程保存的...

为了完整起见，我想提一下 doc of the PUBHandler class

我在 Win7 上使用 python(x,y) 发行版，其中包含 python 2.7.10 和 pyzmq 14.7.0-14

[更新]我排除了 Windows 防火墙作为丢失包的来源

Answer 1

问题出在订阅者一方。最初，订阅者过滤掉所有消息，直到设置过滤器。使用 socket.setsockopt(opt, value) 函数来存档它。pyZMQ对这个函数的使用描述不是很清楚:

getsockopt(opt) get default socket options for new sockets created by this Context

但是zmq_setsockopt函数的文档非常清楚(see here):

int zmq_setsockopt (void *socket, int option_name, const void *option_value, size_t option_len)

...

ZMQ_SUBSCRIBE: Establish message filter The ZMQ_SUBSCRIBE option shall establish a new message filter on a ZMQ_SUB socket. Newly created ZMQ_SUB sockets shall filter out all incoming messages, therefore you should call this option to establish an initial message filter.

因此，解决方案是使用 socket.setsockopt(zmq.SUBSCRIBE,filter) 设置过滤器，其中 filter 是您要过滤的字符串。使用 filter='' 显示所有消息。像 filter='ERROR' 这样的过滤器只会显示错误消息并抑制所有其他类型，例如 WARNING、INFO 或 DEBUG。

这样，sub_client() 函数如下所示:

import time
import zmq

def sub_client():
    port = "12344"       
    context = zmq.Context()
    socket = context.socket(zmq.SUB)    
    socket.connect("tcp://127.0.0.1:%s" % port)
    socket.setsockopt(zmq.SUBSCRIBE,'')


    # Process 30 updates
    print "Listening to publishers..."
    for i in range (30):       
        print "Listening to publishers..."
        message = socket.recv()
        print "Received error #%s: %s" % (str(i), message)
        time.sleep(1)

sub_client()