ZMQ pub/sub 用于传输 base64 图像答案

【问题标题】：ZMQ pub/sub for transfering base64 imagesZMQ pub/sub 用于传输 base64 图像
【发布时间】：2019-12-22 15:59:30
【问题描述】：

我面临一个任务：使用 zmq 套接字发送和接收 base64 字符串（从 800x600 图像生成）。目前，我正在使用 pub/sub 连接来执行此任务。但是看起来消息很大以至于套接字无法立即传输它，并且后面的消息卡在网络缓冲区中。虽然我不想丢失这么多消息，但我必须限制 HWM 值，以便套接字正常工作。所以我有一些问题：

还有其他有效的库/方法来执行我的任务吗？还是应该使用 zmq 提供的其他连接类型（路由器/经销商和请求/回复）？
要传输图像（由 OpenCV 处理的图像），除了转换为 base64 格式外，我是否可以使用一种方法来最小化发送图像的大小？
如果我必须继续使用 zmq pub/sub 连接，我该如何限制存储旧消息的时间，而不是它们的数量，比如 3 分钟？

这是我的套接字 python 代码：

发布者

import numpy as np
import zmq
import base64
context = zmq.Context()
footage_socket = context.socket(zmq.PUB)
footage_socket.setsockopt(zmq.SNDHWM, 1)
footage_socket.connect(<tcp address>)

def send_func(frame, camera_link):
   height, width, _ = frame.shape
   frame = np.ascontiguousarray(frame)
   base64_img = base64.b64encode(frame)
   data = {"camera_link":camera_link,'base64_img':base64_img, "img_width":width, "img_height":height}
   footage_socket.send_json(data)

订阅者

footage_socket = context.socket(zmq.SUB)
footage_socket.bind(<tcp address>)
footage_socket.setsockopt(zmq.RCVHWM, 1)
def rcv_func(): 
    while True:
        print("run socket")
        try:
            framex = footage_socket.recv_string()
            data = json.loads(framex)
            frame = data['base64_img']
            img = np.frombuffer(base64.b64decode(frame), np.uint8)
            img = img.reshape(int(frame_height), int(frame_width), 3)

        except Exception as e:
            print(e)

【问题讨论】：

为什么需要base64？为什么不只是字节？ docs.scipy.org/doc/numpy/reference/generated/…
对不起，最近我们只尝试了这种方法。我会试试你的建议
好的，告诉我进展如何。
我不确定元数据对您有多重要（高度、宽度等）。你也可以尝试pickle numpy 数组。
我看到如果我将 python 函数 sys.getsizeof() 与 numpy 数组一起使用，它返回的值小于 base64 字符串。但我不确定消息的大小如何影响发送和接收字段的速度

标签： python sockets image-processing base64 zeromq

【解决方案1】：

_{在开始之前，让我做一些说明：

- 如果只是为了便于编码，请避免将数据重新打包到 JSON 中。 JSON-re-serialised 数据“增长”的大小，没有为您提供超快速和资源高效的流处理的单一增值。只有当专业系统有足够的时间和几乎无限的空闲 CPU 处理能力时，他们才会“求助” JSON 格式，他们浪费在将有价值的数据重新打包到另一个数据盒中-inside-another-box-of-data。在可行的情况下，他们可以支付所有成本和效率低下 - 在这里，您将导致什么都得不到花费的 CPU 时钟，重新打包所需的 RAM 增加一倍以上，并且还必须传输更大的数据

- 回顾，如果相机确实提供了“值得”成为 8 字节 / 64 位的图像数据deep”，如果没有，您将获得第一个出色的免费图像数据缩减}

使用sys.getsizeof() 可能会让你大吃一惊：

>>> aa = np.ones( 1000 )
>>> sys.getsizeof(  aa )
8096 <---------------------------- 8096 [B] object here "contains"
>>> (lambda array2TEST: array2TEST.itemsize * array2TEST.size )( aa )
8000 <---------------------------- 8000 [B] of data
>>> bb = aa.view()     # a similar effect happen in smart VECTORISED computing
>>> sys.getsizeof( bb )
96 <------------------------------   96 [B] object here "contains"
>>> (lambda array2TEST: array2TEST.itemsize * array2TEST.size )( bb )
8000 <---------------------------- 8000 [B] of data
>>> bb.flags
  C_CONTIGUOUS    : True
  F_CONTIGUOUS    : True
  OWNDATA         : False <-------------------------------||||||||||
  WRITEABLE       : True
  ALIGNED         : True
  WRITEBACKIFCOPY : False
  UPDATEIFCOPY    : False
  >>> bb.dtype
  dtype('float64') <--------------    8 [B] per image-pixel even for {1|0} B/W

问：有没有一种方法我可以使用最小化发送图像的大小...？ em>

是的，已经花费了数百万 [man * years] 的研发资金，致力于解决这个问题，并且仍在不断发展最好的解决方法。

最好的结果，正如任何人自己可能已经预料到的那样，对于极端极端的情况是需要的 - 用于从远方、遥远的深空、回家的卫星图像传输 - 就像 JAXA 在它的第二次运行时一样@ 987654321@，这次是visiting the Ryugu asteroid。

您的原样代码以迄今为止未指定的 fps 速率和颜色深度生成 800x600 图像帧。一个简短的视图显示，如果处理过程没有得到更多的关注和应有的注意，那么在上述 -3 分钟- 内可以轻松生成多少数据：

>>> (lambda a2T: a2T.itemsize * a2T.size )( np.ones( ( 800, 600, 3 ) ) ) / 1E6
11.52 <---- each 800x600-RGB-FRAME handled/processed this way takes ~ 11.5 [MB]
                    @~30 fps                                        ~345.6 [MB/s]
                                                                    ~ 62.2 [GB/3min]

解决方案？从一流的专业知识中汲取灵感：

你有有限的功率（无论是在能源方面还是在处理方面 - 不要忘记，这颗卫星“内部”的 CPU-s 已经制造了大约 5 - 7 年以前，在项目启动之前 - 没有人敢用明亮而炙手可热的新但未经证实的 COTS 芯片发送任务），有限 RAM（再次，功率加重量限制，作为数量起飞和“那里”飞行所需的燃料随着每一克“有用的有效载荷”而增长），最后但并非最不重要的 - 最限制因素 - 你有非常有限的 R/F-COMM 手段 - 一个如此“looooooooong”的线（如果您尝试从“这里”回答任何遥控器，则需要将近半天的时间才能从“那里”回到“这里”+同样的情况- 检测到错误后请求或请求重新发送）。当前的 DSN 有效遥测数据传输速度约为 6.4 ~ 9.6 kbps（是的，不超过约 7000 位/秒）

在这里，最聪明的头脑已经将人类智慧的所有艺术投入到实现这一目标中：

图像压缩的终极手段 - 除非确实重要且必要，否则切勿发送任何信息
转码图像数据的终极手段错误自我纠正添加 - 如果有什么值得添加的，错误检测不是（你将不得不等待几乎一天，让它再次“重新传输”，希望那里没有另一个错误）。在这里我们需要一种（有限的——见上面发送单个比特的成本，所以这必须是非常经济的附加）自我纠正的方法，它确实可以修复一些有限范围的信号/数据传输错误，在从深空返回家园的 R/F-COMM 信号期间可能会出现并且确实会出现。对于较大的错误，您必须等待几天才能通过另一次尝试发送更大的数据包来解决重新安排的图像数据错误恢复，这是无法通过内置功能从“损坏”数据中恢复的- 错误自我纠正。

从哪里开始？

如果您的用例没有大量可用的“备用” CPU 能力（确实需要有足够的“空闲” CPU+RAM 资源来执行任何此类高级图像数据转码和错误恢复重新处理，无论是在规模（体积用于转码和重新处理的附加数据 - 两者都具有大尺寸 - 数量级大于单个图像帧的大小）和时间（附加 CPU 的速度-processing ) ) 获得最终的图像数据压缩没有什么神奇的技巧，您的故事到此结束。

如果您的用例可以提高 CPU 能力，那么您的下一个敌人就是时间。设计一个足够聪明的图像处理的时间和处理每个图像帧的时间，使用您设计的图像数据转码，在相当短的时间内，在发送到接收端之前。前者可由您的项目资源管理（由财务 - 让合适的熟练工程师加入，并由执行（执行）实际设计和工程阶段的人员）。后者是不可管理的，它取决于您项目的需求 - 您的项目仍然可以存活以执行预期功能的速度 (fps) 和承受的延迟（延迟的累积 [ms] 多晚）。

python 是一个简单的原型生态系统，一旦你需要提高吞吐量（参考上面），这很可能（30 多年的经验让我非常有信心这么说 - 即使你加入附加类固醇，比如进入 cython + C-extensions 来完成整个马戏团确实有点，但只是快一点，必须付出巨大的附加成本（获得新技能如果尚未加入 - 拥有昂贵的学习曲线持续时间并且那些熟练的人的薪水会增加）重新设计和重构您迄今为止良好原型的代码库）将是该节目的第一个障碍
OpenCV 可以并且将为您提供一些基本的图像处理工具，让您从
图像数据转码和普通或终极数据压缩必须遵循，以减少数据大小
ZeroMQ 是问题最少的部分 - 性能方面的可扩展性和独特的低延迟吞吐能力。如果没有任何细节，人们可能会忘记 PUB/SUB，除非您一直阻止和避免任何订阅列表处理（这样做的成本会对 {central- node | network-dataflows + all remote-nodes }-重载，对预期的快速和适当大小的图像数据管道处理没有实际影响。

Q : 如果我必须继续使用 zmq pub/sub 连接，我该如何限制存储旧消息的时间，而不是它们的数量，比如 3 分钟？

ZeroMQ 是一款智能工具，但必须了解它的强大功能 - ZeroCopy 将帮助您在生产中保持低 RAM 配置，但 如果您计划存储 -3 分钟的图像数据流，您将需要大量的 RAM 和 CPU 能力，而且这在很大程度上还取决于 .recv()-ing 对等节点的实际数量。

ZeroMQ 是一个无代理系统，因此您实际上并不“存储”消息，但.send()-方法只是告诉 ZeroMQ 基础设施，只要 ZeroMQ 基础设施提供的数据是免费发送的看到有机会将它们发送给指定的对等接收者（无论是在本地还是在大西洋或通过卫星连接）。这意味着，正确的 ZeroMQ 配置是必须的，如果您计划让发送/接收端准备好对最压缩的图像数据进行入队/传输/接收/出队约 3 分钟流，可能会提供多个流，以防 1：多方通信出现在生产中。

正确的分析和合理的设计决策是您的项目能够满足所有这些要求的唯一机会，因为 CPU、RAM 和传输装置是先验已知的有限。

【讨论】：