ZeroMQ：SUB 可以订阅的主题数量有限制吗？

Question

我正在使用 ZeroMQ 的 PUB/SUB 套接字模式。 PUB 制作股票的财务数据并发布。主题设置为每只股票的代码。在 SUB 端，客户可以根据股票代码订阅他们想要的数据。 PUB 用 C 编写，SUB 用 Python 编写。

但是，在测试过程中出现了问题。如果只将一个股票代码设置为 SUB 套接字上的消息过滤器，则一切正常。但是当涉及到大量股票时，程序会在短时间内崩溃，并报错“Segmentation fault (core dumped)”（详情如下）。

这是PUB (C)的代码：

while (1) {  
    int rc = 0;
    // send topic
    rc = zmq_send(pub_socket, topic, rc, ZMQ_SNDMORE);
    if (rc == -1) {
        // error handling
    }
    // send stock data
    rc = zmq_send(pub_socket, data, rc, 0);
    if (rc == -1) {
        // error handling
    }
}

这里是SUB (Python)的代码：

import zmq

# initialize a SUB socket
ctx = zmq.Context.instance()
socket = ctx.socket(zmq.SUB)

# set socket options to filter message
for code in code_list:
    socket.setsockopt_string(zmq.SUBSCRIBE, code)

socket.connect(PUB_ADDR)
# recv data from PUB
while True:
    data = socket.recv()
    print(data)

我还使用 gdb 来调试程序。 调试结果如下： debug result

有人知道程序为什么会崩溃吗？欢迎任何帮助，谢谢。

---

更新：

如果我将setsockopt_string 部分替换为以下代码，Python 脚本运行良好。奇怪的事情...我需要深入挖掘setsockopt_string 函数。

Python 中的新代码：

socket.setsockopt_string( zmq.SUBSCRIBE, "" )

---

最新更新：

我运行了@user3666197 提供的脚本，得到了调试日志。我只选择了日志的几个部分，因为它很长。

套接字初始化 setsockopt_string 完成 接收一个味精并退出

Answer 1

简介：

PUB-side 使用 ZeroMQ v 4.1.5； SUB-side 使用 ZeroMQ Python wrapper 16.0.2

这隐含地使PUB/SUB 模式（与返回到 v 2.0 的前几代 API 形成对比）依赖于 PUB 端过滤，而您的 SIGSEGV 指示在SUB-side 上报告问题。

尽管假设过滤是根本原因，但我记得一些关于大树过滤问题的技术辩论，仍然有一个小惊喜，就像在一些关于 Trie-search 的帖子中一样，添加了 "" 叶节点也做了一个神奇的服务。如果有帮助，将尝试再次找到这场辩论。

在 ZeroMQ 过滤器中初始 remarks from Martin Sustrik refer up to 约 10,000 个订阅不会产生问题（在进一步的设计讨论中会有一些更高的数字）：

高效订阅匹配

在 ZeroMQ 中，简单的尝试用于存储和匹配 PUB/SUB 订阅。订阅机制适用于最多 10,000 个订阅，其中简单的 trie 运行良好。但是，有些用户使用了多达 150,000,000 个订阅。在这种情况下，需要更有效的数据结构。因此，nanomsg 使用内存高效版本的 Patricia trie 而不是简单的 trie。

更多详情请查看this article。

---

始终至少使用分步方法诊断原因：

轻微的测试修改将使您更接近于打开问题的真实信封：

import zmq
pass;                                           print "DEBUG: Ok, zmq imported. [ver:{0:}]".format( zmq.pyzmq_version() )
#_______________________________________________# SETUP ZMQ:
ctx    = zmq.Context( 2 )                       # Context( nIOthreads )
pass;                                           print "DEBUG: Ok, zmq.Context() instantiated."  
socket = ctx.socket(  zmq.SUB )                 # Socket(  .SUB )
pass;                                           print "DEBUG: Ok, Socket instantiated."
socket.connect(           PUB_ADDR )            # .connect()
pass;                                           print "DEBUG: Ok, .connect() completed."
socket.setsockopt(   zmq.LINGER, 0 )            # explicit LINGER
pass;                                           print "DEBUG: Ok, .setsockopt( LINGER, 0 ) completed."
#_______________________________________________# SET FILTER:
for code in code_list:
    pass;                                       print "DEBUG: Going to set SUB side n-th filter: {0: > 1000d}. == [{1:}]".format( code_list.index( code ), repr( code ) ),
    socket.setsockopt_string( zmq.SUBSCRIBE, code )
    pass;                                       print "DEBUG: Ok, this one was done."
pass;                                           print "DEBUG: Ok, all items from <code_list> exhausted."
#_______________________________________________# LOOP FOREVER:
while True:
    try:
          print "LOOP: .recv() call."
          data = socket.recv()
          print "LOOP: .recv()-ed {0:}[B] repr()-ed as [{1:}]".format( len( data ), repr( data ) )

    except KeyboardInterrupt():
          print "EXC: Ctrl-C will terminate."

    except:
          print "EXC: will terminate."

    finally:
          pass;                                 print "DEBUG: Ok, finally: section entered:"
          socket.close()
          pass;                                 print "DEBUG: Ok, Socket instance .close() call returned"
          ctx.term()
          pass;                                 print "DEBUG: Ok, .Context() instance term()-ed"
          break

鉴于描述为只有一个 PUB 和一个 SUB 的测试用例，其他性能扩展和详细的缓冲区管理问题不会爆发目前的问题。运行修改后的测试并发布琐碎的 DEBUG:log 后，将看到结果。

每秒发送大约 3k 条消息也不成问题。

---

更新：漏点 -- (1) Unicode 处理 + (2) 主题过滤器

(1) 如 DEBUG:log 中所示，您可以混合使用 Unicode 和纯 C 字节数组。 These representations MUST match - system-wide（从.send_string()，通过.setsockopt_string()，直到.recv_string()）

data = socket.recv_string()           # AS YOUR DEBUG:log shows the b'mkt_bar...'

(2) 主题-过滤器必须匹配 - 否则消息将被分类为未订阅的消息 ... 这样 u'abc .... '过滤器匹配 u'abc ....' 消息。否则：

setsockopt_string( option, optval, encoding='utf-8' )

长度为零的空 optval 应订阅所有传入消息。 非空optval 应订阅所有以指定前缀开头的消息。 多个过滤器可以附加到单个 ZMQ_SUB 套接字，在这种情况下，一条消息应该是如果匹配至少一个过滤器，则接受。

DEBUG：上面提供的日志片段显示（嗯，PrintScreens ... -- 请下次复制/粘贴终端 ASCII，而不是图片，除非显示一些 GUI 功能，对吗？谢谢... )，您的主题过滤器在定义的意义上永远不会匹配。修复它。系统范围。

ZeroMQ 不应该因此受到责备，Unicode + C 字节数组根本无法工作，而且如果混合使用或调用接口错误。

---

结语：

如果仍然指责 ZeroMQ 主题过滤能力，最简单的 (dis) 批准 Null 假设的 a/b 测试将是运行 VERY SAME 测试，但仅使用 5-topic-filter 元素。如果这也崩溃了，那么您关于容量相关限制的假设是错误的。

继续走！