将相同数据发送到多个动态进程的有效方法答案

【问题标题】：Efficient way of sending the same data to multiple dynamic processes将相同数据发送到多个动态进程的有效方法
【发布时间】：2020-05-11 20:23:02
【问题描述】：

我有一个行缓冲数据流，以及来自其他进程的许多读取器
读取器需要动态附加到系统，写入流的进程不知道它们

首先我尝试阅读每一行并将它们简单地发送到很多管道

#writer
command | while read -r line; do
    printf '%s\n' "$line" | tee listeners/*
done

#reader
mkfifo listeners/1
cat listeners/1

但这会消耗大量 CPU

所以我想写一个文件并反复清理它

#writer
command >> file &
while true; do
    : > file
    sleep 1
done

#reader
tail -f -n0 file

但有时，一行在截断之前没有被一个或多个读取器读取，从而产生竞争条件
有没有更好的方法来实现这一点？

【问题讨论】：

command | tee listeners/* 你为什么要绕行？ that's consume a lot of CPU哇，为什么？你是怎么测量的？通常的方法是在 posix 兼容的系统上创建管道，在 bash 上通常的方法是使用command | tee >(listener1) >(listener2) | listener3
可以创建新的监听器，但不会更新通配符扩展
需要缓冲行吗？ IE。如果有新的听众出现，他是否需要读取所有数据？您应该使用while IFS= read -r line，否则它将删除前导和尾随空格。可能会很慢 - 对于每一行，您正在运行 3 个进程 read printf 和 tee。
您用“shell”标记了这个问题，这意味着您想要仅 posix 的解决方案？如果您使用bash 进行标记，您的问题将会接触到更多人。
鉴于对性能的关注（由 CPU 利用率和缺少的行暗示） - 如果您提供输入（频率、大小、消息/秒）和消费者（这些脚本是？程式？）。看起来一个简单的解决方案行不通 - 其他详细信息将有助于提出解决方案

标签： shell system

【解决方案1】：

对我来说听起来像pub/sub - 请参阅Wikipedia。

基本上，新的感兴趣的人会随时出现并“订阅”到您的频道。接收数据的进程然后将其“发布”逐行到该通道。

您可以使用 mosquitto 或 Redis 使用 MQTT。两者都有命令行接口/绑定，以及 Python、C/C++、Ruby、PHP 等。客户端和服务器不必在同一台机器上，一些客户端可以在网络上的其他地方。

蚊子示例here。

我在 Mac 上使用 Redis pub/sub 进行了一些测试。终端中用于订阅名为myStream 的频道的客户端代码如下所示：

redis-cli SUBSCRIBE myStream

然后我运行一个过程来合成 10,000 行，如下所示：

time seq 10000  | while read a ; do redis-cli PUBLISH myStream "$a" >/dev/null 2>&1 ; done

这需要 40 秒，因此每秒大约 250 行，但它必须为每一行启动一个全新的进程，并创建和断开与 Redis 的连接......我们不想发送你的CPU 疯了。

那么，更适合您的情况的是，您可以使用以下方法创建一个包含 100,000 行的文件，一次读取一个文件，然后用 Python 将它们发送给您的所有订阅者：

# Make a "BigFile" with 100,000 lines
seq 100000 > BigFile

并阅读这些行并将其发布：

#!/usr/bin/env python3

import redis

if __name__ == '__main__':
    # Redis connection
    r = redis.Redis(host='localhost', port=6379, db=0)

    # Read file line by line...
    with open('BigFile', 'r') as infile:
        for line in infile:
            # Publish the current line to subscribers
            r.publish('myStream', line)

整个 100,000 行在 4 秒内发送和接收，因此每秒 25,000 行。这是它在行动中的一点记录。在顶部你可以看到 CPU 并没有受到它的过度困扰。从上往下的第二个窗口是客户端，接收 100,000 行，下一个窗口是第二个客户端。底部窗口显示服务器运行上面的 Python 代码并在 4 秒内发送所有 100,000 行。

关键字：Redis、mosquitto、pub/sub、发布、订阅。

【讨论】：