multithreading - 线程::队列什么都不返回

Question

我在下面的脚本中工作，在该脚本中我将文件内容分割为@file_splitted并尝试应用Thread::Queue来加快过程。但是$result最后没有返回任何内容。你能检查一下发生了什么吗？

my $NUM_WORKERS = 5;

my $q = Thread::Queue->new();

sub worker {
   my ($job) = @_;

   print "@_  \n################\n";
   my ($sub_name, @args) = @$job;
   my $sub_ref = \&Subroutine;
   $sub_ref->(@args);
}

{
    my $q = Thread::Queue->new();

    my @workers;

    for (1..$NUM_WORKERS) {
        push @workers, async {
            while (my $job = $q->dequeue()) {
            worker($job);
#           print "$job \n";
            }
        };
    }

    $q->enqueue($_) for @file_splitted;
    $q->end();

    for my $t(@workers){
        (my @getit)= $t->join();
        my $tmp = join '', @getit; 
        $result .= $tmp;
        print "$result\n";
    }

}

Answer 1

这是您当前的代码，但是整理了一下，并进行了注释:

my $NUM_WORKERS = 5;

my $q = Thread::Queue->new();

{
    # ok, so here we create a new queue for some reason that shadows the outer $q
    my $q = Thread::Queue->new();

    my @workers;
    for (1 .. $NUM_WORKERS) {
        push @workers, async {
            while (my $job = $q->dequeue()) {
                my ($sub_name, @args) = @$job; # so the $job is an arrayref
                Subroutine(@args);  # what is "Subroutine"?
            }
            # this worker does not explicitly return any values
        };
    }

    # what is @file_splitted? Does it contain arrayrefs?
    $q->enqueue($_) for @file_splitted;
    $q->end();

    for my $t (@workers){
        # where is $result declared? And why are you using a return value when you
        # don't explicitly return anything from your threads?
        $result .= join '', $t->join;
        print "$result\n";
    }

}

问题是您实际上并没有从线程中返回任何有用的信息–请注意，我删除了上面的 worker子例程，因为它没有为该讨论添加任何内容，并且可能使您感到困惑。

您很可能希望创建另一个队列，线程可以从该队列返回结果:

my $job_queue = Thread::Queue->new;
my $result_queue = Thread::Queue->new;
my @workers;
for (1 .. $NUM_WORKERS) {
    push @workers, async {
        while(defined(my $job = $job_queue->dequeue)) {
            my $result = Subroutine($job);  # or something like this
            $result_queue->enqueue($result);
        }
        $result_queue->enqueue(undef);
    };
}

$job_queue->enqueue(@jobs);
$job_queue->end;

my $waiting = $NUM_WORKERS;
my @results;
while ($waiting) {
    if (defined(my $result = $result_queue->dequeue)) {
        # do something with the results
        push @results, $result;
    }
    else {
        $waiting--;
    }
}

$_->join for @workers;

如果只想最后收集所有结果，则可以执行以下操作:

my $job_queue = Thread::Queue->new;
my $result_queue = Thread::Queue->new;
my @workers;
for (1 .. $NUM_WORKERS) {
    push @workers, async {
        while(defined(my $job = $job_queue->dequeue)) {
            my $result = Subroutine($job);  # or something like this
            $result_queue->enqueue($result);
        }
    };
}

$job_queue->enqueue(@jobs);
$job_queue->end;

$_->join for @workers;

my @results = $result_queue->dequeue($result_queue->pending);