Verständnis async in Perl auf spezifisches Beispiel

Question

Ich muss ein Skript schreiben, die einige URLs parallel erhalten und etwas Arbeit erledigen. In der Vergangenheit habe ich immer Parallel::ForkManager für solche Dinge verwendet, aber jetzt wollte ich etwas Neues lernen und versuchen asynchrone Programmierung mit AnyEvent (und AnyEvent::HTTP oder AnyEvent::Curl::Multi) ... aber ich habe Probleme, AnyEvent zu verstehen und ein Skript zu schreiben, das sollte:

• öffnen einer Datei (jede Zeile ist eine separate URL)
• (von nun parallel, aber mit einer Grenze für zB 10 gleichzeitige Zugriffe)
• Lesedatei Zeile für Zeile (ich möchte nicht laden ganze Datei in den Speicher - es könnte groß sein)
• eine HTTP-Anfrage machen f oder dass URL
• Leseantwort-
• Updates MySQL Datensatz entsprechend
• (nächste Datei Zeile)

Ich habe viele Handbücher lesen, Tutorials, aber es ist immer noch schwer für mich Unterschiede zwischen Sperrung und nicht zu verstehen -Blockierungscode. Ich habe ähnliches Skript bei http://perlmaven.com/fetching-several-web-pages-in-parallel-using-anyevent gefunden, wo Herrn Szabo die Grundlagen erklärt, aber ich kann immer noch nicht verstehen, wie so etwas implementieren:

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

... und eine Gleichzeitigkeit Grenze in diesem Fall hinzufügen.

Ich würde für Hilfe sehr dankbar;)

UPDATE

Nach Ikegami Rat, den ich Net::Curl::Multi einen Versuch gab. Ich bin sehr zufrieden mit den Ergebnissen. Nach Jahren der Verwendung Parallel::ForkManager nur für die gleichzeitige Erfassung von Tausenden von URLs scheint Net::Curl::Multi genial zu sein. Hier ist mein Code mit while Schleife auf Dateihandle. Es scheint zu funktionieren, wie es sollte, aber da ich zum ersten Mal so etwas schreibe, möchte ich erfahrenere Perl-Benutzer bitten, einen Blick darauf zu werfen und mir zu sagen, ob es ein paar potentielle Fehler gibt, etwas, das ich vermisste usw. wenn ich fragen darf: Da ich nicht ganz verstehe, wie die Parallelität von Net::Curl::Multi funktioniert, bitte sagen Sie mir, ob ich irgendwelche Probleme mit MySQL UPDATE Befehl (über DBI) innerhalb RESPONSE Schleife erwarten sollte (neben höheren Serverlast offensichtlich - ich erwarte final Skript, um mit ungefähr 50 gleichzeitigen N::C::M Arbeitern zu laufen, vielleicht mehr).

#!/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 ist eine ziemlich gute Bibliothek, die extrem schnell ist. Darüber hinaus kann es auch parallele Anfragen bearbeiten! Ich würde empfehlen, dies anstelle von AnyEvent zu verwenden.

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

Dies macht genau das, was Sie wollen, in einer asynchronen Art und Weise, und es tut, dass Net::Curl auf sichere Art und Weise durch Umwickeln:

#!/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;