在 Perl 中 fork 多个子节点并使用管道进行双向通信答案

【问题标题】：Fork multiple children in Perl and use pipes for bi-directional communication在 Perl 中 fork 多个子节点并使用管道进行双向通信
【发布时间】：2012-01-15 10:07:55
【问题描述】：

我正在尝试创建一个具有多处理能力的小型 Perl 程序。由于我的要求中到处都有一些小的更改，因此我无法在任何地方找到任何类似的示例脚本。

我需要从 STDIN 读取一个大的日志文件，然后给第一个子进程的前 N 行（又是一个大的数字），然后给第二个子进程等接下来的 N 行。我还定义了一个常量这是允许同时运行的最大子进程数。一旦达到孩子的最大数量，父母将等待孩子完成它的工作并给它另外 N 行。

父进程还收集每个子进程完成时返回的多行（5-10 行）输出并将其存储在数组中。然后Parent继续处理这个数组内容并最终显示结果。

是否有更好的示例脚本可供我修改和使用，或者有人可以在这里分享一个帮助我吗？我更喜欢只使用管道进行进程间通信，并尽可能让事情变得更简单。

编辑：有人可以举例说明如何仅使用 IO::Handle 模块中的管道来完成此操作吗？

【问题讨论】：

标签： perl fork pipe

【解决方案1】：

使用Forks::Super，这样可以轻松限制并发进程的数量并处理进程间通信。例如，

use Forks::Super MAX_PROC => 10,     # allow 10 simultaneous processes
                 ON_BUSY => 'queue'; # don't block when >=10 jobs are active

@loglines = <>;

# set up all the background jobs
while (@loglines > 0) {
    $pid = fork {
        args => [ splice @loglines, 0, $N ],  # to pass to sub, below
        child_fh => "out",    # make child STDOUT readable by parent
        sub => sub {
            my @loglines = @_;
            my @result = ... do something with loglines ...
            print @results;   # use $pid->read_stdout() to read in child
        }
    };
}

# get the results
while ($pid = waitpid -1, 0) {
    last if $pid == -1;
    my @results_from_job = $pid->read_stdout();
    push @results, @results_from_job;
}

【讨论】：

看起来正是我想要的。但是我在一个有限的环境中运行这个脚本，那里只有有限数量的 Perl 模块可用。我们可以单独使用 IO::Handle 并达到同样的效果吗？

【解决方案2】：

我发现线程对于这种过程要简单得多。您需要线程和 Threads::Queue 模块。该过程是设置一个队列来为工作线程提供数据，并为它们返回结果。工作线程只是一个读取记录、处理它并返回结果的函数。我只是把这段代码放在一起并没有测试它，所以它可能是错误的，但我认为显示了一般的想法：

use threads ();
use Thread::Queue;
#
#
#            Set limit on number of workers
#
my $MAX_THREADS = 5;
my $EOD = "\n\n";
#
#
#            Need a queue to feed the workers
#            and one for them to return results
#
my $Qresult  = Thread::Queue->new();
my $rec;
my $n;
#
#
#            load STDIN into the input queue
#
my $Qin = Thread::Queue->new(<>);
#
#
#            start worker threads
#
for($n = 0; $n < $MAX_THREADS; ++$n)
{
    async{ProcessRecord($n);};
    $Qin->enqueue($EOD);            #    need terminator for each
}
#
#
#
#            Wait for the results to come in
#
$n = 0;
while($n < $MAX_THREADS)
{
            $rec = $q->dequeue();
            if($rec eq $EOD)
            {
                ++$n;
                next;
            }

            :
            :
            :
    #-- process result --#
            :
            :
            :


    threads->yield();    #    let other threads get a chance
    sleep 1;
}
exit;    
######################################    
#
#
#            Worker threads draw from the queue
#            when a "terminator" is read, quit;
#
sub ProcessRecord 
{
    my $rec;
    my $result;

    while(1)
    {
        $rec = $Qin->dequeue();
        last if $rec eq $EOD;
            :
            :
            :
        #-- process record --#
            :
            :
            :
        $Qresult->enqueue($result);

        threads->yield();    #    let other threads get a chance
    }

    threads->exit();    
}

【讨论】：