当前位置：文江博客话题详情

ZeroMQ 持久化模式

发布于 2024-09-30 02:34:07 字数 104 浏览 1 评论 0原文

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

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

我想知道使用 ZeroMQ 的模式。

原文

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

时常饿 2024-10-07 02:34:07

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

回复收藏 0 原文

假情假意假温柔 2024-10-07 02:34:07

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

持久性方面将消息保留一段时间（http://en.wikipedia.org/wiki/Time_to_live），这是为了给客户端提供连接的机会。我使用了内存数据结构，但也考虑过使用 Redis 来获得磁盘持久性。

回复收藏 0 原文

翻身的咸鱼 2024-10-07 02:34:07

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

该模块位于：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()

We needed to persist the received messages from a subscriber before processing them. The messages are received in a separate thread and stored on disk, while the persisted message queue is manipulated in the main thread.

The module is available at: https://pypi.org/project/persizmq. From the documentation:

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

回复收藏 0 原文

~没有更多了~