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

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)

结果

^{pr2}$

日志订户：

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()的调用时被锁定。我在不同的控制台上启动了publisher和subscriber。当我使用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上使用了python2.7.10和pyzmq14.7.0-14的python（x，y）发行版

[更新] 我排除了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或{}。在

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

import time
import zmq

def sub_client():
    port = "12344"       
    context = zmq.Context()
    socket = context.socket(zmq.SUB)    
    ssocket.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()

Answer 2

我猜你没有在服务器上设置PUB socket。在

像这样的事情应该可以

publisher = context.socket(zmq.PUB)
publisher.bind('tcp://127.0.0.1:12344')
zmqh = PUBHandler(publisher)
logger = logging.getLogger('myapp')
logger.setLevel(logging.ERROR)
logger.addHandler(zmqh)

我希望这有帮助。在