persizmq 1.0.3

线程化订阅者

线程化订阅者实现为persizmq.ThreadedSubscriber。您需要指定一个回调函数，该函数在接收到每条消息时被调用。

您还需要指定异常回调，以便在监听线程中处理抛出的异常。

示例

import time

import zmq

import persizmq

context = zmq.Context()
subscriber = context.socket(zmq.SUB)
subscriber.setsockopt_string(zmq.SUBSCRIBE, "")
subscriber.connect("ipc:///some-queue.zeromq")

def callback(msg: bytes)->None:
    print("received a message: {}".format(msg))

def on_exception(exception: Exception)->None:
    print("an exception was raised in the listening thread: {}".format(exception))

with persizmq.ThreadedSubscriber(callback=callback, subscriber=subscriber, on_exception=on_exception):
    # do something while we are listening on messages...
    time.sleep(10)

存储

我们为接收到的消息提供两种存储模式

1. persizmq.PersistentStorage：在磁盘上以FIFO队列的形式存储消息。

2. persizmq.PersistentLatestStorage：仅存储最新的消息在磁盘上。

存储组件直接作为回调传递给线程化订阅者。

示例

import pathlib

import zmq

import persizmq

context = zmq.Context()
subscriber = context.socket(zmq.SUB)
subscriber.setsockopt_string(zmq.SUBSCRIBE, "")
subscriber.connect("ipc:///some-queue.zeromq")

persistent_dir = pathlib.Path("/some/dir")
storage = persizmq.PersistentStorage(persistent_dir=persistent_dir)

def on_exception(exception: Exception)->None:
    print("an exception was raised in the listening thread: {}".format(exception))

with persizmq.ThreadedSubscriber(callback=storage.add_message, subscriber=subscriber, on_exception=on_exception):
    msg = storage.front()  # non-blocking
    if msg is not None:
        print("Received a persistent message: {}".format(msg))
        storage.pop_front()

过滤

我们还提供可以链式添加到线程化订阅者的过滤组件。如果您只想持久化少量消息并忽略其余部分，过滤链尤其有用。

过滤器实现于persizmq.filter模块。

示例

import pathlib

import zmq

import persizmq
import persizmq.filter

context = zmq.Context()
subscriber = context.socket(zmq.SUB)
subscriber.setsockopt_string(zmq.SUBSCRIBE, "")
subscriber.connect("ipc:///some-queue.zeromq")

persistent_dir = pathlib.Path("/some/dir")
storage = persizmq.PersistentStorage(persistent_dir=persistent_dir)

def on_exception(exception: Exception)->None:
    print("an exception was raised in the listening thread: {}".format(exception))

with persizmq.ThreadedSubscriber(
    lambda msg: storage.add_message(persizmq.filter.MaxSize(max_size=1000)(msg)),
    subscriber=subscriber,
    on_exception=on_exception):

    msg = storage.front()  # non-blocking
    if msg is not None:
        print("Received a persistent message: {}".format(msg))
        storage.pop_front()