在 Perl 中分叉多个子项并使用管道进行双向通信

Question

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

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

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

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

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

Answer 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;
}

Answer 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();    
}