PERL - 多线程 - 子 + 孙 - 运行线程的限制

Question

我制作了一个 Perl 脚本，在其中创建线程（同时运行的线程受到限制），每个线程创建自己的子节点，这些子节点的数量也应该受到限制。

在我托管我的脚本的地方，我不能同时为每个 Perl 脚本启动超过 X 个线程。在下面的示例中，我同时最多有 X = 3 x 7 = 21 个线程。

• 第一份工作 3 ($nb_process_first)
• 第二份工作 7 ($nb_process_second)

问题：

1. 是否有更好的方法来管理线程及其子项？ （例如队列 - 请给我一些代码示例，因为我尝试过但没有成功）
2. 尽管我在所有正在运行的线程上使用循环来加入它们（参见脚本末尾），但我当前的脚本并未终止并加入所有线程。

#!/usr/bin/perl -s

use threads;

my @threads;
my $nb_process_first = 3;
my @running          = ();

print "START" . "\n";

$current = 1;
while ( $current <= 10 ) {
    @running = threads->list(threads::running);
    if ( scalar @running < $nb_process_first ) {
        print "Launch firstJob=" . scalar @running . "\n";
        my $thread = threads->create( \&firstJob );
        push( @threads, $thread );
    } else {
        redo;
    }
    $current++;
}

my @joinable = threads->list(threads::joinable);
while ( scalar @joinable != 0 ) {
    foreach my $thr ( threads->list() ) {
        $thr->join();
    }
    @joinable = threads->list(threads::joinable);
}

print "END" . "\n";

sub secondJob {
    for ( $i = 0; $i <= 15; $i++ ) {
        print "secondJob=" . $i . "\n";
        sleep 1;
    }
    threads->exit();
}

sub firstJob {
    my $nb_process_second = 7;
    my @running           = ();
    $current = 1;
    while ( $current <= 10 ) {
        @running = threads->list(threads::running);
        if ( scalar @running < $nb_process_second ) {
            print "firstJob/Launch secondJob=" . scalar @running . "-" . $current . "\n";
            my $secondthread = threads->create( \&secondJob );
            push( @threads, $secondthread );
            sleep 2;
        }
        $current++;
    }
    threads->exit();
}

Answer 1

Thread::Queue 是线程代码的基本“工作线程”模型的便捷模型。

有点像这样：

#!/usr/bin/perl
use strict;
use warnings;
use threads;
use Thread::Queue;

my $firstworkitem_q = Thread::Queue -> new();
my $secondworkitem_q = Thread::Queue -> new();
my $nthreads = 10; 

sub first_worker {
   while ( my $item = $firstworkitem_q -> dequeue() ) {
       print "First worker picked up $item, and queues it to second worker\n";
       $secondworkitem_q -> enqueue ( $item ); 
   }
}

sub second_worker {
   while ( my $item = $secondworkitem_q -> dequeue() ) {
       print "Second worker got $item"; 
   }
}

my @first_workers;
for ( 1..$nthreads ) {
    my $thr = threads -> create ( \&first_worker );
    push ( @first_workers, $thr );
}

for ( 1..$nthreads ) {
    my $thr = threads -> create ( \&second_worker );
}

$firstworkitem_q -> enqueue ( @things_to_processs );
$firstworkitem_q -> end;

foreach my $firstworker ( @first_workers ) {
   $firstworker -> join();
}

#here all the first workers have finished, so we know nothing will be queued to second work queue. 
$secondworkitem_q -> end();

foreach my $thr ( threads -> list() ) {
    $thr -> join();
}

你把东西塞进队列，然后迭代处理。当您 end 队列时，while 循环将获得 undef 并因此终止 - 使您的线程可连接。

您不需要像您那样跟踪@running，因为threads -> list() 会这样做。更重要的是 - 你需要使 @running 成为一个共享变量并锁定它，否则你在每个线程中都有它的不同副本。

拥有firstJob spawn secondJob 我会避开，因为它会产生各种水果虫。我建议产生两类工作线程。使用$queue -> end() 触发第一组工人关闭。

Answer 2

至于你的第二个问题，线程只有在它们完成运行后才能加入（参见this answer）。因为在第二个while 循环运行时某些线程没有完成，所以它在没有加入它们的情况下完成。

您的循环应该根据活动线程的数量而不是可连接线程的数量来等待。像这样的：

while (threads->list() > 0)
{
   foreach my $joinable (threads->list(threads::joinable))
   {
       $joinable->join();
   }
}

至于第一个问题，当然还有其他管理线程的方法。但是，在不了解您的任务的情况下，不可能说出您应该做什么。