(py)zmq/PUB :是否可以调用connect()，然后立即发送()而不丢失消息？

Question

有了这段代码，我总是会失去这样的信息：

def publish(frontend_url, message):
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.connect(frontend_url)
    socket.send(message)

然而，如果我引入一个简短的睡眠()，我可以得到这样的信息：

def publish(frontend_url, message):
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.connect(frontend_url)
    time.sleep(0.1)  # wait for the connection to be established 
    socket.send(message)

是否有一种方法可以确保消息不会在连接()和发送()的调用之间休眠而传递？

恐怕我无法预测睡眠时间(网络潜伏期等)。

更新：

上下文:我想将数据更新从烧瓶 REST应用程序发布到message (例如。关于资源的创建/更新/删除)。

目前，消息代理是使用0mq转发器设备起草的。

我知道0mq是为抽象TCP套接字和消息传递复杂性而设计的。

在连接时间很长的情况下，我可以用它。然而，当我在像gunicorn或uwsgi这样的应用程序容器中运行我的Flask应用程序时，我有N个工作进程，我不能期望连接或进程会持续很长时间。

正如我所理解的那样，我应该使用真正的messages (如RabbitMQ)，并使用同步客户端在那里发布消息。

Answer 1

你不能完全做到这一点，但是也许还有其他的解决方案可以解决你的问题。

为什么要使用PUB/SUB套接字？pub/sub的特性更适合长期运行的套接字，通常您将在bind()套接字上使用PUB，并在SUB套接字上进行连接。您在这里所做的，是旋转一个套接字来发送一条消息，大概是发送到某种类型的“服务器”，并不完全符合PUB/SUB范式。

如果您选择REQ或DEALER的某些变体到REP或ROUTER，那么您的事情可能会变得更顺利。REQ套接字将保存一条消息，直到它的对子准备好接收消息为止。如果您不特别关心来自“服务器”的响应，那么您可以直接放弃它。

您不只是打开套接字，而不是构建一个全新的上下文和套接字，并在每次发送消息时重新连接，有什么特别的原因吗？我可以想到一些有限的场景，其中这可能是首选的行为，但一般来说，最好不要使用套接字。如果你想坚持使用PUB/SUB，那么在你的应用程序开始时就把套接字旋转起来，睡一段安全的时间，包括任何合理的延迟场景，然后开始发送你的消息，而不用担心每次重新连接。如果您要在没有任何新消息的情况下长期使用这个套接字，您可能需要使用心跳来确保连接保持打开。

Answer 2

来自ZMQ指南

关于PUB子套接字还有一件更重要的事情要知道:您不知道订阅者何时开始接收消息。即使启动订阅服务器，等待一段时间，然后启动发布服务器，订阅服务器也始终会忽略发布服务器发送的第一条消息。这是因为当订阅服务器连接到发布服务器(需要较小但非零的时间)时，发布者可能已经在发送消息。

Answer 3

这里的许多帖子都是从以下几个方面开始的：

“我用的是.PUB/.SUB，它没有我想要它做的工作.这里的任何人，一定要帮我把它做好，就像我认为它会开箱而出一样。”

这种方法在现实世界中行不通，在分布式系统设计中越少，在系统中就越差，在这些系统中，几乎实时的调度和/或严格的资源管理是无法避免的。

进程间/跨平台消息传递并不是“另一个”简单代码行(SLOC)。

# A sample demo-code snippet          # Issues the demo-code has left to be resolved
#------------------------------------ #------------------------------------------------
def publish( frontend_url, message ): # what is a benefit of a per-call OneStopPUBLISH function?
    context = zmq.Context()           # .Context() has to be .Terminate()-d (!)
    socket = context.socket(zmq.PUB)  # is this indeed "a disposable" for each call?
    socket.connect(frontend_url)      # what transport-class used for .connect()/.bind()?
    time.sleep(0.1)                   # wait for the connection to be established 
    socket.send(message)              # ^ has no control over low-level "connection" handshaking

任何人都可以起草几个一行，并投入相当大的努力(自己或社区外包)，以使它最终发挥作用(至少在某种程度上)。

然而，这是一个具有巨大能力的领域，因此需要稍微重塑一个人的头脑，以使其潜力得到开放和充分利用。

勾勒出一个好的解决方案的需要，但有错误的理由或错误理解的SLOC-s (不管是否复制/粘贴-d)，通常不会产生任何对近期合理的东西，对于更远的未来就越少。

消息传递只是引入了一种新的模式--一个新的宏观宇宙--更大范围的楼宇自动化--令人惊讶的是，你的(确定性)代码成为了一组更复杂的有限自动机( Finite State Automata，FSA )的成员，这并不奇怪，因为我们打算做一些“信息传递”--相互交流。

为此，需要一些地方资源管理，一些“外部”交通，一些“正式行为模式礼仪”(而不是相互呼喊)的沟通-原始。

这通常内置在ZeroMQ、nanomsg和其他库中。

然而，有两件重要的事情仍然是隐藏的。

• 事物内部运作方式的微观宇宙(许多人，如果不是全部的话，试图调整这一点，而不是最好地利用它，通常都是浪费时间)。
• 如何将一群其他琐碎的元素(通信原语)编排成一个健壮的、可扩展的消息传递体系结构、跨进程/本地主机/网络边界进行合作并满足总体设计需求的macro-cosmos。

无法理解这两个世界之间的距离，通常会导致我们在消息库中没有充分利用我们预先准备好的最大优势。

简单地说，最好的做法就是忘记一条线的调整方法。这是没有成效的。

首先了解全局观点，让你能够利用那些对你的好处最有效的力量来实现你的目标。

为什么这么复杂？

​

​

(由nanomsg.org提供)

任何非平凡的系统都是复杂的。无论是在TimeDOMAIN还是在ResourcesDOMAIN。如果一个人努力创建一个稳定的、智能的、高性能的、低延迟的、与传输类无关的通用通信框架，那么它就越多.

好消息是，这已经被详细阐述，并内置到微宇宙架构中。

坏消息是，这并不能从盒子里解决你的需求(除了一些非常琐碎的事情)。

这里我们来看看宏观宇宙的设计。

你的责任是设计一个更高空间的算法，如何使许多孤立的FSA原语进行转换，并找到一个与不断发展的多到多对话相一致的协议。是。库为您提供了“公正”的原始构建块(非常强大，毫无疑问)。但让“外层空间”满足你的需要是你的责任。

这可能而且通常是复杂的。

嗯，如果这是微不足道的，那么它很可能已经包括在“里面”的图书馆，不是吗？

下一个要去哪里？

也许下一步最好的办法是让IMHO朝着更全面的方向迈出一步，这对于尝试用ZeroMQ编码的最初几件事情来说可能也会很复杂，但如果您至少跳到Pieter的书，代码连接，第1卷的265页，如果不是一步一步地阅读的话。

人们可以开始认识到，如何开始对FSA基元的宏观宇宙进行“编程”，从而形成一个更高层次的FSA-FSA-FSA，它能够并将解决所有ProblemDOMAIN特定的问题。

首先，在图60重新发布更新和图62克隆服务器对上有一个未公开的视图，然后只尝试返回根、元素和细节。