zeromq 持久性模式答案

【问题标题】：zeromq persistence patternszeromq 持久性模式
【发布时间】：2011-05-02 20:41:48
【问题描述】：

谁必须管理 ZeroMQ 中的持久化？

当我们在 Python 语言中使用 ZeroMQ 客户端时，有哪些插件/模块可用于管理持久化？

我想知道使用 ZeroMQ 的模式。

【问题讨论】：

ZeroMQ 指南有一个称为 Titanic 的持久性模式。它基于一种名为 MajorDomo 的模式。您还可以很容易地创建自己的持久队列。但是，在开始添加持久性之前，有必要详细了解您的可靠性要求。《指南》对此进行了深入介绍。仅在某处添加持久性不会给您带来可靠性。
@PieterHintjens 请将此添加为答案。我差点错过了。它是此页面上最有价值的资源。

【解决方案1】：

据我所知，Zeromq 没有任何持久性。它超出了它的范围，需要由最终用户处理。就像序列化消息一样。在 C# 中，我使用 db4o 来添加持久性。通常我将对象保持在其原始状态，然后将其序列化并将其发送到 ZMQ 套接字。顺便说一句，这是 PUB/SUB 对的。

【讨论】：

【解决方案2】：

在应用程序端，您可以相应地持久化，例如，我在 node.js 中构建了一个持久性层，它与后端 php 调用和通过 websockets 进行通信。

持久性方面将消息保存一段时间（http://en.wikipedia.org/wiki/Time_to_live），这是为了给客户端一个连接的机会。我使用了内存中的数据结构，但我玩弄了使用 redis 来获得磁盘持久性的想法。

【讨论】：

【解决方案3】：

我们需要在处理订阅者收到的消息之前对其进行持久化。消息在单独的线程中接收并存储在磁盘上，而持久消息队列在主线程中操作。

该模块位于：https://pypi.org/project/persizmq。来自文档：

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 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()

【讨论】：