Zrozumienie asynchronizacji w perlu na konkretnym przykładzie

Question

Muszę napisać skrypt, który pobiera kilka adresów URL równolegle i wykonuje pewną pracę. W przeszłości zawsze używałem Parallel::ForkManager takich rzeczy, ale teraz chciałem nauczyć się czegoś nowego i spróbować programowania asynchronicznego z AnyEvent (i AnyEvent::HTTP lub AnyEvent::Curl::Multi) ... ale mam problem ze zrozumieniem AnyEvent i napisaniem skryptu, który powinien:

• otworzyć plik (każdy wiersz jest oddzielna URL)
• (od teraz równolegle, ale z ograniczeniem do karm 10 jednoczesnych żądań)
• odczytu pliku linia po linii (nie chcę załadować cały plik do pamięci - może być duży)
• utworzyć żądanie HTTP f albo że URL
• odpowiedź odczytu
• aktualizacje rekord MySQL odpowiednio
• (następna linia plik)

Czytałem wiele podręczniki, samouczki, ale to nadal trudno mi zrozumieć różnice między blokowania i non blokujący kod. Znalazłem podobny scenariusz w http://perlmaven.com/fetching-several-web-pages-in-parallel-using-anyevent, gdzie pan Szabo wyjaśnia podstawy, ale ja wciąż nie mogę zrozumieć, jak zaimplementować coś takiego:

... 
open my $fh, "<", $file; 
while (my $line = <$fh>) 
{ 
# http request, read response, update MySQL 
} 
close $fh 
... 

... i dodać limitu współbieżności w tym przypadku.

Byłbym bardzo wdzięczny za pomoc;)

UPDATE

Za radą Ikegami za dałem Net::Curl::Multi spróbować. Jestem bardzo zadowolony z wyników. Po latach używania Parallel::ForkManager tylko dla równoczesnego zbierania tysięcy adresów URL, Net::Curl::Multi wydaje się być niesamowity. Oto mój kod z pętlą while na uchwycie pliku. Wydaje się działać tak, jak powinno, ale biorąc pod uwagę, że po raz pierwszy piszę coś takiego, chciałbym poprosić bardziej doświadczonych użytkowników Perla o sprawdzenie i powiedzenie, czy są jakieś potencjalne błędy, coś, za czym tęskniłem itp. . jeśli mogę zapytać: ponieważ nie rozumiem w pełni, jak działa współbieżność Net::Curl::Multi, proszę powiedz mi, czy powinienem spodziewać się problemów z wprowadzeniem polecenia UPDATE do MySQL (przez DBI) wewnątrz pętli RESPONSE (poza oczywiście większym obciążeniem serwera - spodziewam się ostatecznego skrypt uruchamiany z około 50 współbieżnymi pracownikami N::C::M, może więcej).

#!/usr/bin/perl 

use Net::Curl::Easy qw(:constants); 
use Net::Curl::Multi qw(); 

sub make_request { 
    my ($url) = @_; 
    my $easy = Net::Curl::Easy->new(); 
    $easy->{url} = $url; 
    $easy->setopt(CURLOPT_URL,  $url); 
    $easy->setopt(CURLOPT_HEADERDATA, \$easy->{head}); 
    $easy->setopt(CURLOPT_FILE,  \$easy->{body}); 
    return $easy; 
} 

my $maxWorkers = 10; 

my $multi = Net::Curl::Multi->new(); 
my $workers = 0; 

my $i = 1; 
open my $fh, "<", "urls.txt"; 
LINE: while (my $url = <$fh>) 
{ 
    chomp($url); 
    $url .= "?$i"; 
    print "($i) $url\n"; 
    my $easy = make_request($url); 
    $multi->add_handle($easy); 
    $workers++; 

    my $running = 0; 
    do { 
     my ($r, $w, $e) = $multi->fdset(); 
     my $timeout = $multi->timeout(); 
     select $r, $w, $e, $timeout/1000 
     if $timeout > 0; 

     $running = $multi->perform(); 
     RESPONSE: while (my ($msg, $easy, $result) = $multi->info_read()) { 
      $multi->remove_handle($easy); 
      $workers--; 
      printf("%s getting %s\n", $easy->getinfo(CURLINFO_RESPONSE_CODE), $easy->{url}); 
     } 

     # dont max CPU while waiting 
     select(undef, undef, undef, 0.01); 
    } while ($workers == $maxWorkers || (eof && $running)); 
    $i++; 
} 
close $fh; 

Answer 1

Net :: Curl to dość dobra biblioteka, która jest niezwykle szybka. Ponadto może obsługiwać również żądania równoległe! Zalecam używanie tego zamiast AnyEvent.

use Net::Curl::Easy qw(:constants); 
use Net::Curl::Multi qw(); 

sub make_request { 
    my ($url) = @_; 
    my $easy = Net::Curl::Easy->new(); 
    $easy->{url} = $url; 
    $easy->setopt(CURLOPT_URL,  $url); 
    $easy->setopt(CURLOPT_HEADERDATA, \$easy->{head}); 
    $easy->setopt(CURLOPT_FILE,  \$easy->{body}); 
    return $easy; 
} 

my $max_running = 10; 
my @urls = ('http://www.google.com/'); 

my $multi = Net::Curl::Multi->new(); 
my $running = 0; 
while (1) { 
    while (@urls && $running < $max_running) { 
     my $easy = make_request(shift(@urls)); 
     $multi->add_handle($easy); 
     ++$running; 
    } 

    last if !$running; 

    my ($r, $w, $e) = $multi->fdset(); 
    my $timeout = $multi->timeout(); 
    select($r, $w, $e, $timeout/1000) 
     if $timeout > 0; 

    $running = $multi->perform(); 
    while (my ($msg, $easy, $result) = $multi->info_read()) { 
     $multi->remove_handle($easy); 
     printf("%s getting %s\n", $easy->getinfo(CURLINFO_RESPONSE_CODE), $easy->{url}); 
    } 
} 

Answer 2

To robi dokładnie to, co chcesz, w sposób asynchroniczny, a robi to poprzez owinięcie Net::Curl w bezpieczny sposób:

#!/usr/bin/env perl 

package MyDownloader; 
use strict; 
use warnings qw(all); 

use Moo; 

extends 'YADA::Worker'; 

has '+use_stats'=> (default => sub { 1 }); 
has '+retry' => (default => sub { 10 }); 

after init => sub { 
    my ($self) = @_; 

    $self->setopt(
     encoding   => '', 
     verbose    => 1, 
    ); 
}; 

after finish => sub { 
    my ($self, $result) = @_; 

    if ($self->has_error) { 
     print "ERROR: $result\n"; 
    } else { 
     # do the interesting stuff here 
     printf "Finished downloading %s: %d bytes\n", $self->final_url, length ${$self->data}; 
    } 
}; 

around has_error => sub { 
    my $orig = shift; 
    my $self = shift; 

    return 1 if $self->$orig(@_); 
    return 1 if $self->getinfo('response_code') =~ m{^5[0-9]{2}$}x; 
}; 

1; 

package main; 
use strict; 
use warnings qw(all); 

use Carp; 

use YADA; 

my $q = YADA->new(
    max  => 8, 
    timeout => 30, 
); 

open(my $fh, '<', 'file_with_urls_per_line.txt') 
    or croak "can't open queue: $!"; 
while (my $url = <$fh>) { 
    chomp $url; 

    $q->append(sub { 
     MyDownloader->new($url) 
    }); 
} 
close $fh; 
$q->wait;