假设有两个应用程序，其中应用程序A使用发布者客户端不间断地向应用程序B进行数据流传输，而应用程序B具有一个子服务器套接字来接受该数据，我们如何配置应用程序A中的pub客户端套接字，以便当B不可用（例如它正在重新部署、重新启动）时，A缓存所有未决消息，并且当B变为可用时，缓冲的消息会通过并使套接字追上实时流？
简而言之，我们如何使PUB客户端套接字在SUB服务器不可用时缓冲一定量的消息？
PUB客户端的默认行为是进入静默状态，但如果我们可以将其更改为限制大小的缓冲区，那将非常好，这在zmq中是否可能？还是我需要在应用程序级别上进行处理...
我尝试在我的套接字中设置HWM和LINGER，但如果我没有错的话，它们只负责慢消费者情况，其中我的发布者连接到订阅者，但订阅者太慢，以至于发布者开始缓冲消息（hwm将限制这些消息的数量）...
我正在使用jeromq，因为我针对jvm平台。
首先，欢迎来到Zen-of-Zero的世界，延迟最为重要。
ZeroMQ由Pieter HINTJENS团队设计，其中最有经验的大师Martin SUSTRIK被命名为第一位。该设计专业精良，以避免任何不必要的延迟。因此，问及是否具有（有限的）持久性？不，先生，未经确认的“PUB/SUB可扩展正式通信模式原型”将不会内置它，因为这会增加问题并降低性能和可伸缩性（添加延迟，处理和内存管理）。
如果需要（有限的）持久性（用于不存在远程SUB端代理的连接），请随意在应用程序端实现它，或者可以设计并实现一个新的符合ZMTP标准的行为模式原型，扩展ZeroMQ框架，如果这项工作进入稳定和公开接受的状态，但不要请求高性能，削减延迟的标准PUB/SUB朝这个方向进行修改。这绝对不是正确的做法。
使用双指针循环缓冲区，工作在一种类似于Persistence-PROXY的(app-side-managed)模式下，App端可以轻松实现您添加的逻辑，但位于PUB发送方之前。
如果您的设计还喜欢使用最近提供的内置ZeroMQ-socket_monitor组件来设置额外的控制层并在那里接收事件流，则您的设计可能会成功地从ZeroMQ内部细节中挤出一些额外的内容。这些与网络和连接管理相关的事件可以为您的(app-side-managed)-Persistence-PROXY带来更多的启示。
然而，请注意：
所以如果计划使用最终有趣的传输类之一，就可以直接忘记这个
_zmq_socket_monitor()_方法仅支持面向连接的传输，即TCP、IPC和TIPC。
{ inproc:// | norm:// | pgm:// | epgm:// | vmci:// }。
我们的社区荣誉成员smac89提供的信息存在不准确、甚至错误的部分，他尽力回答了您在评论中表达的额外兴趣：
"...zmq 优化了主题发布吗？比如如果你快速地在一些 100 字符长的
topic上发布，它实际上是每次都发送topic吗，还是将其映射到某个 int 并随后发送 int...?"
他告诉你：
"它将始终发布
topic。当我使用pub-sub模式时，我通常先发布topic，然后再发布实际消息，所以在订阅者中，我只需读取第一帧并忽略它，然后读取实际消息"
ZeroMQ 不是这样工作的。没有“单独”的<topic>，后面跟着一个<message-body>，相反是相反的
TOPIC 和主题过滤的机械化工作方式非常不同。
1) 你永远不知道，谁会调用 .connect()：
例如，可以几乎确定版本2.x到版本4.2+将以不同的方式处理主题过滤（ZMTP：RFC定义了初始能力版本握手，让Context-实例决定使用哪个版本的主题过滤：
ver 2.x 通常将所有消息移动到所有对等方，并让所有SUB-side（来自ver 2.x+）接收消息（并让SUB-side Context-实例处理本地topic列表过滤处理）
而
ver 4.2+ 确定在**PUB-side Context-实例上执行topic-列表过滤处理（CPU使用率增加，网络传输相反），因此您的SUB-side将永远不会收到任何“无用”的字节读作“未订阅”消息。
2) (你可以这样做，但是)没有必要将“主题”分成因此暗示的多帧消息的第一帧。也许恰恰相反（在高性能、低延迟的分布式系统设计中这样做是一种反模式）。
主题过滤过程被定义为逐字节工作，从左到右，对于主题列表成员值，进行与传递的消息有效负载的每个模式匹配。
添加额外数据，额外的帧管理处理只会增加端到端的延迟和处理开销。这不是一个好主意，而是应该进行适当的分布式系统设计工作。
在专业的分布式系统设计中，没有容易的胜利，也没有低垂的果实，特别是当低延迟或超低延迟是设计目标时。
另一方面，确保ZeroMQ框架考虑到了这一点，并且这些努力得到了稳定、终极性能良好、平衡的一套工具，用于智能（按设计）、快速（运行）和可扩展（魔鬼也会羡慕）的信令/消息服务，人们喜欢使用它们，正因为这种设计智慧。
希望您享受ZeroMQ的使用，并随意在您选择的应用程序套件内添加任何额外的功能“前端”。
我在这里发布一个快速更新，因为其他两个答案（虽然很有用但事实上是错误的），我不希望其他人从我的被接受的答案中获得错误信息。 不仅可以使用zmq来完成此操作，而且它实际上是默认行为。
诀窍是，如果您的发布者客户端之前从未连接到订阅服务器，则会不断丢弃消息（这就是为什么我认为它不会缓冲消息的原因），但是如果您的发布者连接到订阅者并且您重新启动订阅者，则发布者将缓冲消息直到达到HWM，这正是我所要求的...因此简而言之，发布者想知道有没有人在另一端接收消息，只有在这之后它才会缓冲消息...
下面是一些演示这一点的示例代码（您可能需要进行一些基本编辑才能编译它）。
我只使用了这个依赖项org.zeromq:jeromq:0.5.1。
zmq-publisher.kt
fun main() {
val uri = "tcp://localhost:3006"
val context = ZContext(1)
val socket = context.createSocket(SocketType.PUB)
socket.hwm = 10000
socket.linger = 0
"connecting to $uri".log()
socket.connect(uri)
fun publish(path: String, msg: Msg) {
">> $path | ${msg.json()}".log()
socket.sendMore(path)
socket.send(msg.toByteArray())
}
var count = 0
while (notInterrupted()) {
val msg = telegramMessage("message : ${++count}")
publish("/some/feed", msg)
println()
sleepInterruptible(1.second)
}
}
当然，还有zmq-subscriber.kt
fun main() {
val uri = "tcp://localhost:3006"
val context = ZContext(1)
val socket = context.createSocket(SocketType.SUB)
socket.hwm = 10000
socket.receiveTimeOut = 250
"connecting to $uri".log()
socket.bind(uri)
socket.subscribe("/some/feed")
while (true) {
val path = socket.recvStr() ?: continue
val bytes = socket.recv()
val msg = Msg.parseFrom(bytes)
"<< $path | ${msg.json()}".log()
}
}
先尝试运行发布者（publisher），不开启订阅者（subscriber）。然后当你启动订阅者时，你错过了到目前为止所有的消息...现在，不需要重新启动发布者，停止订阅者并等待一段时间，然后再次启动它。
这是我一个服务实际受益于此的示例...以下是结构：[当前服务]sub:server <= pub:client[被重启的服务]sub:server <=* pub:client[多个发布者]
因为我在中途重启了服务，所以所有的发布者都开始缓存它们的消息，最终的服务观察到每秒约200条消息的下降至0（其中1或2条是心跳包），然后突然出现1000多条消息的爆发，因为所有的发布者都刷新了它们的缓冲区（重新启动大约需要5秒钟）......在这里我实际上没有失去任何一条消息...
请注意，你必须有subscriber:server <= publisher:client配对（这样发布者就知道“只有一个地方需要将这些消息传递给”（你可以尝试在发布者上绑定并在订阅者上连接，但你将不再看到发布者缓冲消息，因为它的可疑性是：刚刚断开连接的订阅者是否停止需要数据或者是它出现故障了）。
Socket.monitor并监听ZMQ_EVENT_CONNECTED和ZMQ_EVENT_DISCONNECTED，因为这些事件会告诉您客户端何时连接/断开连接，从而使您能够切换到填充所选缓冲区的操作。当然，可能有其他方法可以实现此功能，而不直接涉及zeromq，但这取决于您自己的决定。sendMore 发送主题，然后用 send 发送 flatbuffer 字节。在订阅方面，我使用 recv 接收主题，然后下一个带有标志零的 recv 给出 flatbuffer。我想知道的是，我在这里做错了什么，不符合 zmq 的方式，我该如何解决？ - smac89
socket.monitor并监听ZMQ_EVENT_DISCONNECTED事件吗？ - smac89zeromqHWM docs，它说：“当您的套接字达到其HWM时，它将根据套接字类型阻止或丢弃数据。PUB和ROUTER套接字如果达到其HWM，则会丢弃数据，而其他套接字类型则会阻塞。在inproc传输中，发送方和接收方共享相同的缓冲区，因此真正的HWM是由双方设置的HWM之和。”所以我的问题是，您是否正在寻找一个外部内存缓冲区作为解决方案？除了我的建议之外，使用大HWM是唯一的解决方案。 - smac89