使用 ZeroMQ 关闭套接字时如何检测是否达到 linger？答案

【问题标题】：How to detect whether linger was reached when closing socket using ZeroMQ?使用 ZeroMQ 关闭套接字时如何检测是否达到 linger？
【发布时间】：2020-12-13 08:28:21
【问题描述】：

以下dispatch() 函数运行通过Queue.queue 接收消息，并使用ZeroMQ PUSH 套接字将它们发送到端点。

我希望这个函数退出，一旦它通过队列接收到None，但是如果套接字的底层消息缓冲区有任何未传递的消息（远程端点关闭），那么应用程序将不会终止。因此，一旦函数接收到None，它就会关闭带有指定延迟的套接字。

使用这种方法，我如何检测是否达到了指定的逗留时间？特别是，不会引发异常。

def dispatch(self):    
    context = zmq.Context()
    socket = context.socket(zmq.PUSH)
    poller = zmq.Poller()

    socket.connect('tcp://127.0.0.1:5555')
    poller.register(socket, zmq.POLLOUT)

    while True:
        try:
            msg = self.dispatcher_queue.get(block=True, timeout=0.5)
        except queue.Empty:
            continue

        if msg is None:
            socket.close(linger=5000)
            break

        try:
            socket.send_json(msg)
        except Exception as exc:
            raise common.exc.WatchdogException(
                f'Failed to dispatch resource match to processor.\n{msg=}') from exc

【问题讨论】：

标签： python zeromq

【解决方案1】：

Q：“如何检测使用 ZeroMQ 关闭套接字时是否达到linger？”

嗯，这不是一件容易的事。

ZeroMQ 在内部从用户级代码中隐藏了所有这些细节，因为 API （从以往到最近的 v4.3）都是使用 Zen-of-Zero 艺术的所有优点精心打造的，以最大限度地提高性能，_几乎线性缩放和最小延迟。做不支持的零步（如果违反则少）。

解决这个问题可能有三个主要的攻击方向：

可以尝试配置和使用 zmq_socket_monitor() 的事件观察叠加层来分析可达到的最低详细级别的实际事件序列
也可以尝试一种相当粗暴的方法来为zmq.Socket()-instance 设置无限LINGER 属性并通过在设置数量的（现在）软后发送SIGNAL 直接终止阻塞操作-" linger" 已过期，无论是使用 zmq_timers（[ms]-coarse 定时器/回调实用程序的框架）的 v4.3+ 新功能还是自己的
人们可能更愿意通过“包围”对zmq_ctx_term() 的调用来保持事情清洁并仍然实现目标，根据v4.3 记录的API 将阻止（请注意，它不保证在其他 API 版本来回）。这种方式可以帮助您间接检测实际处于阻塞状态的持续时间，例如：

...
NOMINAL_LINGER_ON_CLOSE = 5000
MASK_a = "INF: .term()-ed ASAP, after {0:} [us] from {1:} [ms] granted"
MASK_b = "INF: .term()-ed ALAP, after {0:} [us] from {1:} [ms] granted"
...
socket.setsockopt( zmq.LINGER, NOMINAL_LINGER_ON_CLOSE ) # ____ BE EXPLICIT, ALWAYS
aClk = zmq.Stopwatch()
aClk.start() #_________________________________________________ BoBlockingSECTION
context.term() # /\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\   BLOCKING.......
_ = aClk.stop() #______________________________________________ EoBlockingSECTION
...
print( (      MASK_a if _ < ( NOMINAL_LINGER_ON_CLOSE * 1000 )
         else MASK_b
         ).format(      _,    NOMINAL_LINGER_ON_CLOSE )
       )
...

我对总是，确实ALWAYS相当明确的宣传是基于两者都看到了从版本中改变的爬行“默认值”到版本（期望以同样的方式继续下去是公平的），所以一个负责任的设计应该，确实总是，强制地重新输入那些我们想要就地的值，因为我们的代码将在我们当前版本和我们，凡人，仍然对我们当前假设的哪一部分将在任何未来版本/修订中保持“默认”保持零保证，并且具有相同的不确定性在我们部署的代码片段的域中将存在什么版本组合（截至 EoY-2020，仍有 v2.1、v2.11、v3.x、v4.x 在某处疯狂运行，所以我们永远不会知道，对吗？）

【讨论】：