How to multithread array processing in Perl?

Question

Perl

Hello!
I don't know where to look for information. Decided to explore multithreading. Prior to this, there was no experience in any language.
Main task:
Take an array of elements, let's say there are 1000 elements. Process in a certain number of threads.
More details:
There is a file with a list of domains. I open it and convert each line into an array element.
There is a handler function that visits this domain and checks the response code. If response = 200, prints it to the console.
I don’t know how to scatter it all over the threads, so let’s say I put 10 threads, and the script processed 10 domains at the same time. Also, if one of the domains was checked before the others, do not wait until all domains are checked, but immediately send a new domain to this thread. In general, you need to somehow track completed threads and constantly add new ones, so that exactly 10 threads always work.
If anyone has an example, or, say, you can write on your knee, I will be grateful!

Answer 1

You can try this code as a basis (using Thread::Queue)

#!/usr/bin/perl

use strict;
use warnings;
use threads;
use Thread::Queue;
use Data::Dumper;

my @myarray = (1 .. 1000);
#print Dumper (\@myarray);die;

my $count = shift || 10;
print "Number of threads: $count\n";

my $q = Thread::Queue->new;
my @threads;
for (0 .. $count - 1)
{
    push @threads, async {
        while (defined (my $f = $q->dequeue))
        {
            some_process ($f);
        }
    };
}

for (@myarray)
{
    $q->enqueue ($_);
}

# Tell workers they are no longer needed.
$q->enqueue (undef) for @threads;

# Wait for workers to end
$_->join for @threads;

print "Complete\n";

1;

sub some_process
{
    my $element = shift;
    my $tid = threads->self->tid;
    #my $count = threads->list (threads::running);
    #print "Running threads: $count\n";
    print "Thread $tid started\n";
    open my $F, '>>', $tid . '.txt';
    print $F 'TID: ', $tid, ', element: ', $element, "\n";
    close $F;
    print "Thread $tid stopped\n";
}

Answer 2

Creating and killing threads is a bad idea, you need to create threads that process the array in a loop until it is empty.
That is, the array is used as a queue.
But streams are a problem, IMHO.

#!/usr/bin/env perl

use strict;
use warnings;

use threads;
use threads::shared;

# Создаём расшареную переменную
my @numbers:shared = (1..100);

# Задаём количество потоков
my $threads = shift || 10;

# Создаём потоки и кладём их обекты в массив
my @threads;
for (1..$threads) {
    push @threads, threads->new(
        sub {
            while (@numbers) {
                # Достаём данные из массива
                # Предварительно заблокировава его для остальных потоков
                my $number;
                {
                    # Блокировка работает только в этом скоупе
                    lock(@numbers);
                    $number = shift(@numbers);
                }
                
                print 'Result: '.($number*10)."\n";
            }
        }
    );
}

# Запускаем потоки привязав их к основному процессу
$_->join for (@threads);

Answer 3

Somewhere saw an example at the heart of AnyEvent.
It's easier not to fool around with multithreading, but to use a message broker thread.
There will be a big backlog for scalability.