multithreading - 在perl中控制在给定时间同时运行的最大线程数

Question

我有一个包含文件列表的数组@arr=(a.txt,b.txt,c.txt);

我正在迭代数组并使用 foreach 循环处理文件；该文件的每一行都将生成一个 sql 并将在 DB 服务器上运行。

我想用文件的每一行创建一个线程并查询数据库。我还想控制一次同时运行的最大线程数。

Answer 1

这只是一个简单的工人模型，一个理想的场景。没问题。

use threads;
use Thread::Queue qw( );

use constant NUM_WORKERS => 5;

sub work {
   my ($dbh, $job) = @_;
   ...
}

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

   my @threads;
   for (1..NUM_WORKERS) {
      push @threads, async {
         my $dbh = ...;
         while (my $job = $q->dequeue()) 
            work($dbh, $job);
         }
      };
   }

   while (<>) {
      chomp;
      $q->enqueue($_);
   }

   $q->enqueue(undef) for 1..@threads;
   $_->join() for @threads;
}

将文件名作为参数传递给脚本，或在@ARGV脚本中分配它们。

local @ARGV = qw( a.txt b.txt c.txt );

Answer 2

您可以使用基于Thread::Pool的系统。或任何基于Boss/Worker模型的系统。

Answer 3

有趣的是，我手动控制要运行的线程数。我使用 Hash of the thread id [code snip] my %thr; #我的线程哈希

$count=1;
$maxthreads=5;

while (shift (@data) {
  $syncthr = threads->create(sub {callfunction here}, {pass variables});
  $tid = $syncthr->tid; #get the thread ID
  $thr{$tid} = $syncthr;
  if ($count >= $maxthreads) {
    threads->yield();
    while (1) { # loop until threads are completed
    $num_run_threads = keys (%thr);
    foreach $one_thread ( keys %thr ) {
       if ($thr{$one_thread}->is_running() ) { # if thread running check for error state
          if ($err = $thr{$one_thread}->error() } {
             [ do stuff here]
          }
          # otherwise move on to next thread check
       } else { # thread is either completed or has error
          if ($err = $thr{$one_thread}->error()) {
            [ check for error again cann't hurt to double check ]
          }
          if ($err = $thr{$one_thread}->join()) {
             print "Thread completed id: [$one_thread]\n";
          }
          delete $thr{$one_thread}; # delete the hash since the thread is no more
          $num_run_threads = $num_run_threads - 1; # reduce the number of running     threads
       }
    } # close foreach loop
    @threads = threads->list(threads::running); # get threads
    if ($num_run_threads < $maxthreads ) {
      $count = $num_run_threads; # reset the counter to number of threads running
      if ( $#data != -1 ) { # check to make sure we still have data
        last; # exit the infinite while loop
      } else {
        if (@threads) {
          next; # we still have threads continue with processing
        } else {
          { no more threads to process exit program or do something else }
        }
      } # end else
    } # end threads running
    } # end the while statement
    #Check the threads to see if they are joinable
    undef @threads;
    @threads = threads->joinable()
    if (@threads) {
      foreach $mthread(@threads) {
        if ($mthreads != 0) {
          $thr->join();
        }
      } #end foreach
    } #end @threads
  } #end the if statement
  $count++; Increment the counter to get to number of max threads to spawn
}

这绝不是一个完整的程序。此外，我已将其更改为非常平淡。但是，我已经成功使用了一段时间。特别是在 OO Perl 中。这对我有用并且有很多用途。我可能会错过更多的错误检查，尤其是在超时的情况下，但我是在线程本身中进行的。顺便说一句，线程实际上是我正在调用的子例程。