Perl 多线程程序

Question

我是 Perl 的新手。我想使用线程编写一个 Perl 脚本。我有几个文件说 20 个文件，并希望使用 5 个线程分 4 批处理这些文件。我正在打印线程号。完成一个批次后，下一批的线程号必须以 1 开头。但不是它创建了 20 个线程。请帮助。我的代码如下：

#!/usr/bin/perl -w
use strict;
use warnings;
use threads;
use threads::shared;

my $INPUT_DIR="/home/Documents/myscript/IMPORTLDIF/";
opendir(DIR, $INPUT_DIR) ;
my @files = grep { /^InputFile/ } readdir DIR;
my $count = @files;
#print "Total Files: $count \n";
my @threads;
my $noofthread = 5;
my $nooffiles = $count;
my $noofbatch = $nooffiles / $noofthread;
#print "No of batch: $noofbatch \n";

my $fileIndex = 0;
my $batch = 1;
while ($fileIndex < $nooffiles) {
    print "Batch: $batch \n";
    for (my $i=0; $i < $noofthread && $fileIndex < $nooffiles ; $i++) {

        my $t = threads->new(\&doOperation, $files[$fileIndex], $i)->join;
        push(@threads, $t);
        $fileIndex++;
        print "FileIndex: $fileIndex \n";
    }
    $batch++;
}

sub doOperation () {
    my $ithread = threads->tid() ;
    print "Thread Index : [id=$ithread]\n" ;
    foreach my $item (@_){
        my $filename = $item;
        print "Filename name: $filename \n";
    }

使用线程队列编辑程序：

    #!/usr/bin/perl -w 
    # This is compiled with threading support

    use strict;
    use warnings;

    use threads;
    use Thread::Queue;

    my $q = Thread::Queue->new(); # A new empty queue

    # Worker thread
    my $INPUT_DIR="/home/Documents/myscript/IMPORTLDIF/";
    opendir(DIR, $INPUT_DIR) or die "Cannot opendir: $!";

    my @thrs = threads->create(\&doOperation ) for 1..5;#for 5 threads
    #my @files = `ls -1 /home/Documents/myscript/IMPORTLDIF/`;
    my @files = grep { /^Input/ } readdir DIR or die "File not present present. \n";
    chomp(@files);

    #add files to queue
    foreach my $f (@files){
    # Send work to the thread
    $q->enqueue($f);
    print "Pending items: " + $q->pending()."\n";
    }

    $q->enqueue('_DONE_') for @thrs;
    $_->join() for @thrs;



    sub doOperation () {
    my $ithread = threads->tid() ;
    while (my $filename = $q->dequeue()) {
     # Do work on $item
    return 1 if $filename eq '_DONE_';
    print "[id=$ithread]\t$filename\n";
   }
    return 1;
    }

Answer 1

您可以使用 Paralel:Queue 并创建 4 个线程并将他们可以处理的项目传递给它们。

To fork or not to fork?

use strict;
use warnings;

use threads;
use Thread::Queue;

my $q = Thread::Queue->new();    # A new empty queue

# Worker thread
my @thrs;
push @thrs, threads->create(\&doOperation ) for 1..5;#for 5 threads
my @files = `ls -1 /tmp/`;chomp(@files);
#add files to queue
foreach my $f (@files){
  # Send work to the thread
  $q->enqueue($f);
  print "Pending items: "$q->pending()."\n";
}
$q->enqueue('_DONE_') for @thrs;
$_->join() for threads->list();



sub doOperation () {
    my $ithread = threads->tid() ;
    while (my $filename = $q->dequeue()) {
      # Do work on $item
      return 1 if $filename eq '_DONE_';
      print "[id=$ithread]\t$filename\n";
    }
    return 1;
}

Answer 2

您正在生成一个线程，然后在生成下一个线程之前等待它完成，每个线程处理一个文件。这就是为什么您看到的线程数与文件数一样多。

my $t = threads->new(\&doOperation, $files[$fileIndex], $i)->join;
                                                             ^^^^--- This will block

试试这样的：

....

# split the workload into N batches
#
while (my @batch = splice(@files, 0, $batch_size)) {
  push @threads, threads->new(\&doOperation, @batch);
}

# now wait for all workers to finish
#
for my $thr (@threads) {
  $thr->join;
}

顺便说一句，Thread::Queue 和 Thread-Pool 可能意味着你想做的工作会有更好的设计。