有没有像 xargs 这样的管道数据而不是使用参数的东西？答案

【问题标题】：Is there something like xargs which pipes data instead of using arguments?有没有像 xargs 这样的管道数据而不是使用参数的东西？
【发布时间】：2020-03-24 10:43:38
【问题描述】：

xargs 如果您有一个通过命令行参数接受其输入的命令，那就太好了，但如果您有一个通过 stdin 接受其输入的命令，它就没有那么有用了。

我所看到的类似情况的建议是使用tee 将行复制到多个下游进程，如下所示：

producer | tee >(consumer0 >out0) >(consumer1 >out1) >(consumer2 >out2) | consumer3 >out3;
cat out* | next-stage-of-pipeline

这有一个缺点，即所有消费者都会收到所有生产的行，并且假设似乎consumer{0..3} 是不同的进程。如果我需要每个消费者都相同，并处理部分输入（例如，并行化一个顺序消费者），那么它就不起作用了。

我正在寻找的是类似xargs 的东西，它允许在同一消费者的多个实例之间拆分工作，然后组合输入并继续处理。输出组合的方式应该模仿xargs，所以不能保证行的顺序，但是两行不会拼接在一起（例如：“Hello Fred”和“Hello George”可能以任何顺序出现，但是你不会看到“HHello GeoFregedello”）。

这样的事情的主要用途是处理大量数据，如果为输入的每一行都启动一个，消费者的启动延迟会很明显。如果consumer 的启动成本很低，我只需将它包装在一个小的shell 脚本中，将它的参数通过管道传递到consumer 并使用xargs 调用它。

至少对于我想到的用例，生产者会从内部服务中获取一堆数据，之后消费者需要转换这些数据并进行一些 API 调用。所以生产者和消费者都将是长时间运行的进程，并行运行 API 调用确实会加快速度。

这样的东西会是理想的，但我一直找不到这样做的东西：

producer | ??? -P20 consumer | next-stage-of-pipeline

是否有提供此功能的命令行工具？

【问题讨论】：

你建议如何分割输入？逐行循环？第一个 1/20 到 consumer0，下一个 1/20 到 consumer1，...？输出如何组合？
理想情况下它是可配置的，实际上循环是最容易实现的，如果存在这样的工具，我期望的可能会可用（我认为这就是 xargs 将输入拆分为论据）。为了在存在大量输入的情况下以合理的方式表现，我希望输出在作为工具标准输出（基本上xargs 的行为方式）
我不明白。你就不能xargs -P20 bash -c 'consumer <<<"$1"' -- 吗？
@KamilCuk 如果consumer 快速启动就可以使用，但如果您正在使用像 Java 这样的语言编写的东西，它既有启动成本，又倾向于运行时间越长越快。
@Morgen 为避免额外的混淆，您能否澄清“生产者”是否“昂贵”（时间、内存、启动缓慢等），以及预期产生的数据量。让消费者/生产者并行运行重要吗？

标签： shell parallel-processing xargs

【解决方案1】：

我认为GNU Paralell 可能会满足您的需求。

producer | parallel consumer | next command in pipeline

编辑：直到我开始玩parallel 之前，我才完全理解这个问题。 Parallel 可以将函数作为消费者，您可以在该函数中使用流。

例如

consumer () 
{ 
    echo "$@" | awk '{print $2}'
}
export -f consumer
for i in {1..30}; do echo "foo $i bar"; done | parallel consumer

编辑 2：

您可以使用--pipe 选项，将数据通过管道传送给消费者。 -q 参数在消费者周围放置 shell 引号：

for i in {1..30}; do echo "foo $i bar"; done | parallel --pipe -q awk '{print $2}'

奖金：

parallel 将在顺序处理器上运行作业。如果你有 8 个核心和 16 个消费者，每个核心将获得 2 个消费者（这是可配置的）
parallel 可以在网络上的多台机器上运行，只要它可以 ssh 给它们而不提示输入密码（例如使用 ssh-agent）。

【讨论】：

我要多读一点，听起来很有希望。

【解决方案2】：

在同一消费者的多个实例之间拆分工作

这可以在xargs 的消费者端完成。

例如：

consumer() {
   consumer$(($RANDOM % 4)) "$@"
}
export -f consumer
producer | xargs -P20 bash -c consumer --

会随机选择一个消费者。

：

consumer() {
   seq 4 |
   xargs -i{} consumer{} "$@"
}
export -f consumer
producer | xargs -P20 bash -c consumer --

将为每个消费者运行输入。

不管怎样，你应该明白了。

对于您的 tee，无需任何临时文件即可轻松完成：

consumer() { sed "s/^/$1: /"; }
producer() { seq 3; }
next-stage-of-pipeline() { sed "s/^/Result: /"; }
producer |
{ tee >(consumer 0 >&10) >(consumer 1 >&11) >(consumer 2 >&12) | consumer 3 >&14 ;} 10>&1 11>&1 12>&1 14>&1 | 
next-stage-of-pipeline

如果你需要前。将每个消费者的输入分成 4 份，仍然很容易做到：

filter() { awk -vN="$1" '(NR + N) % 4 == 1'
producer |
{ tee >(
     filter 1 | consumer 0 >&10
) >(
     filter 2 | consumer 1 >&11
) >(
     filter 3 | consumer 2 >&12
) | 
     filter 4 | consumer 3 >&14 
;} 10>&1 11>&1 12>&1 14>&1 | 
next-stage-of-pipeline

【讨论】：

我认为有一个警告，只有缓冲的写入才能保证在 POSIX 系统上是原子的，并且只有当小于 PIPE_BUF 时（例如，默认情况下不缓冲 stderr）
虽然我喜欢这是一个纯 shell 解决方案，但它不能很好地扩展到大约 20 个消费者（当您必须重复几乎相同的代码时很容易出现拼写错误）
>( ... ) 无法缩放，也无法正常缩放。你可以在awk tho 做管道，所以你可以试试。除此之外，我认为我们将不得不编写自己的工具来做到这一点。但是前。最后一个代码 sn -p 做你想到的？将流拆分为 4 个偶数流并将其并行传递给每个消费者？
是的，它完全符合我的要求。我只是暂缓接受，希望存在已经做到这一点的工具。
我认为可以用纯 bash 编写脚本 - 但处理错误会很痛苦，而且速度会很慢。最好是 C 程序。看起来它应该像前任一样工作。 “具有多个后端的 haproxy” - 您收到一个请求，您将请求转发到多个后端之一，并在您跟踪与后端有多少连接的过程中，等等。我相信我可能会对这样的工具感兴趣，好吧，保持问题的开放性，也许有人会提供这样的工具。

【解决方案3】：

GNU Parallel 可以将标准输入拆分为同一命令：

... | parallel --pipe consumer

默认情况下，它会在 \n 上截断，块大小约为 1 MB。这可以通过--recstart/--recend 和--block 进行更改。

如果输入的是一个文件，这就更快了：

parallel -a bigfile --pipepart --block -1 consumer

这将找到bigfile 的大小并将其拆分为每个CPU 线程的一个块。这是即时完成的 - 因此不会创建临时文件。

https://zenodo.org/record/1146014 的第 9 章详细介绍了这一点。

但是，如果您真的希望将整个输入通过管道传输给不同的消费者，GNU Parallel 也可以做到这一点：

# Pipe 1..10 to consumer-a..z
seq 10 | parallel --pipe --tee 'echo consumer-{}; cat' ::: {a..z}

【讨论】：