Управление максимальным количеством потоков, запущенных одновременно в данный момент времени в Perl
У меня есть массив, который содержит список файлов @arr=(a.txt,b.txt,c.txt);
Я перебираю массив и обрабатываю файлы с помощью цикла foreach; каждая строка файла будет генерировать sql и работать на сервере БД.
Я хочу создать один поток с каждой строкой файла и запросить базу данных. Я также хочу контролировать максимальное количество потоков одновременно.
3 ответа
Вы можете использовать систему на основе Thread::Pool. Или любая система, основанная на модели Босс / Рабочий.
Это просто рабочая модель, идеальный сценарий. Нет проблем.
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 );
Интересно, я вручную контролирую, сколько потоков запустить. Я использую Hash идентификатора потока [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. Это работает для меня и имеет довольно много применений. Возможно, мне не хватает еще нескольких ошибок проверки, особенно с таймаутом, но я делаю это в самом потоке. Что, кстати, поток на самом деле является подпрограммой, которую я вызываю.