zmq 连接到使用子进程打开的脚本

Question

我有一个相当复杂的基于套接字 ( ZeroMQ - REQ/REP ) 的 python 程序，我想通过在同一台机器上运行一个简单的套接字脚本来验证它是否正常工作。

测试脚本是这样的。

import subprocess
import zmq
import json

# ...

for call, response in zip(test_calls, expected_responses):
    p = subprocess.Popen(['python', 'main.py'], stdout=subprocess.PIPE)

    context = zmq.Context()
    socket = context.socket(zmq.REQ)
    socket.setsockopt(zmq.RCVTIMEO, 1000)
    socket.connect("tcp://localhost:8084")

    socket.send_string(json.dumps(call))
    r = json.loads(socket.recv_string())

    assert r == response

    p.terminate()
    socket.close()

（可能值得注意的是，它实际上是在nose2中使用这样的测试实现的，但我觉得这超出了这个问题的范围，并且确实会使样本复杂化。这几乎总结了测试中发生的事情）。

85% 的时间，这会奏效，一切都会过去。嗬嗬！另外 15% 的时间，我在 r = json.loads(socket.recv_string()) 线上得到一个 zmq.error.Again: Resource temporarily unavailable（如果我没有设置 zmq.RCVTIMEO，它就会挂起）。

想知道这是否是一个计时过程（子进程无法及时启动/停止），我在这个地方打了几个 time.sleep() 电话，但它似乎没有做任何事情。

我在套接字部分和pdb'd 之后放置了一个catch，检查了python 子进程的stdout。我在应用程序中有一些 print 语句，每次通过套接字调用和响应时都会打印到 stdout，但它没有收到任何输入，所以 recv 当然会超时。

我以前从未遇到过 zmq 的这类问题，所以我认为这可能与使用这样的子进程有关。有谁知道问题可能是什么以及如何解决？

谢谢。

---

更新：所以看起来进程没有终止（尽管在主应用程序中使用了signal.signal(signal.SIGTERM, close_app) 信号）。这会导致对通过 zmq 进行通信的活动进程的混淆吗？最初调用 p.kill() 而不是 p.terminate() 似乎可以解决问题，尽管它仍然以同样的方式失败了一两次。

---

更新 2： 似乎正在工作，直接调用命令 kill：

subprocess.call(['kill', str(p.pid)])
counter = 0
while p.poll() is None:
    time.sleep(0.1)
    counter += 1
    if counter > 20:
        p.kill()

在大多数情况下，这似乎可以优雅地关闭它。

Answer 1

可能是什么问题

可能与所述子进程内未发布的代码有关，这会导致在子进程强制终止期间启动观察到的行为（包括它的其他资源，在智能且功能非常丰富的多线程中进行管理zmq.Context( n_IO_threads = 1 ) 实例，不在您的视线范围内，仅在有限的先验编码/执行控件内）。

考虑{SIGTERM|SIGKILL|...} 而不是紧急制动，
紧急按钮，
不是分布式系统设计中的明智解决方案

一旦进入分布式系统设计，宁可忘记使用无上下文工具，如SIGTERM 等，但最好将自己的软信号控制平面整合到新设计的分布式系统中基础设施。

这有助于“远程”代理的行为符合此类软信号的实际上下文，并允许执行（在您的完全算法控制下）所有必要的保护、资源清理和预终止职责，从而最终优雅地清理退出。

我在这方面可能听起来很老套，但在您的代码最终将所有 zmq.Context() 实例指示为 @987654326 之前，始终将套接字明确指示为 .close() @。据报道这不是必需的，但是在恕我直言，干净和公平地进行资源处理是分布式系统设计/实现中的一项公平职责。

没有例外，没有借口。

该死的在ZMQ_LINGER 中忘记了零

一个例子，值得一提的是 ZMQ_LINGER 参数的 ZeroMQ API 默认值，如果未设置，则默认值为 0 这意味着，一旦这样的ZeroMQ-socket 实例被指示（显式或隐式）.close() 并且碰巧仍然有一个ZMQ_LINGER == 0，套接字-endpoint 将 BLOCK 直到来自对方缓冲区的所有消息都被传递，这可能会导致您的分布式处理挂起，而没有机会在事后解决这种死锁，如果不是预先解决的话- 正确设置不要永远等待待处理的消息。

较新的pyzmq 文档明确警告不要.destroy() 一个zmq.Context 实例（并盲目地让权威发布的.close()-d 获取套接字.destroy()，这是在自己的代码控制之外）

ctx.destroy( linger = None )
关闭与此上下文关联的所有套接字，然后终止上下文。如果指定了linger，则套接字的LINGER sockopt 将在关闭之前设置。

警告

.destroy 涉及调用zmq_close() ，这不是线程安全的。如果其他线程中有活动的套接字，不能调用_{（SIGTERM & al 很可能会忽略哪个建议，不是吗？）}

所以还有更多理由不依赖SIGTERMdevils 的服务。

正在使用的端口

另外，释放占用的传输类资源需要一些时间。因此，拥有一个刚刚发布 IP:port 的代码并不意味着另一个实例/进程/线程可以直接进入并捕捉相同的端口，而没有一些与 O/S 相关的延迟。而是在这方面检查您的资源重用/释放策略（我敢冒任何在该区域阻塞的风险，并使用一些端口地址池来轮换和FILO-排队，以便至少推迟任何潜在的重用情况，直到合理的与 O/S 相关的延迟完全过期——恕我直言，防止阻塞状态比事后处理阻塞状态的异常要好得多。

.bind() 在.connect() 之前

是另一个这样的问题。一旦您的 subprocess.Popen(...) 启动，O/S 服务启动并让子流程开始自行呼吸需要一些时间。

如果您的第一个进程（已经处于活动状态并正在执行）在派生的子进程实例到达.bind() 之前到达.connect()，分布式系统将阻塞。

设置/拆卸往返时间不能减少到零。资源不是一次性的。使用它会产生一些与系统相关的维护和共享开销。

最后.recv_string() 可能并且确实提出了ZMQError EAGAIN

在某些情况下，本地节点中还没有任何消息准备好通过任何.recv*() 方法检索它，无论是flags = zmq.NOBLOCK 模式下的{.recv|.recv_string|.recv_json|&al}。