Hiểu không đồng bộ trong perl trên ví dụ cụ thể

Question

Tôi phải viết một tập lệnh nhận một số URL song song và thực hiện một số công việc. Trước đây tôi đã luôn sử dụng Parallel::ForkManager cho những thứ như vậy, nhưng bây giờ tôi muốn tìm hiểu điều gì đó mới và thử lập trình không đồng bộ với AnyEvent (và AnyEvent::HTTP hoặc AnyEvent::Curl::Multi) ... nhưng tôi đang gặp sự cố khi hiểu AnyEvent và viết tập lệnh cần:

• mở một tập tin (mỗi dòng là một URL riêng biệt)
• (từ nay song song, nhưng với một giới hạn cho fe 10 yêu cầu đồng thời)
• tập đọc từng dòng (tôi không muốn tải toàn bộ tệp vào bộ nhớ - nó có thể lớn)
• thực hiện yêu cầu HTTP f hoặc URL
• phản ứng đọc
• cập nhật kỷ lục MySQL phù
• (dòng tập tin tiếp theo)

Tôi đã đọc rất nhiều sách hướng dẫn, hướng dẫn, nhưng nó vẫn khó cho tôi để hiểu sự khác nhau giữa chặn và phi mã băm. Tôi đã tìm thấy kịch bản tương tự tại http://perlmaven.com/fetching-several-web-pages-in-parallel-using-anyevent, nơi ông Szabo giải thích những điều cơ bản, nhưng tôi vẫn không thể hiểu được làm thế nào để thực hiện một cái gì đó như:

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

... và thêm một giới hạn đồng thời trong trường hợp này.

Tôi sẽ rất biết ơn sự giúp đỡ;)

CẬP NHẬT

Sau lời khuyên Ikegami của tôi đã Net::Curl::Multi một thử. Tôi rất hài lòng với kết quả. Sau nhiều năm sử dụng Parallel::ForkManager chỉ để lấy hàng nghìn URL đồng thời, Net::Curl::Multi có vẻ là tuyệt vời. Đây là mã của tôi với vòng lặp while trên filehandle. Dường như đây là lần đầu tiên tôi viết một cái gì đó như thế này, tôi muốn hỏi người dùng Perl có kinh nghiệm hơn để xem và cho tôi biết nếu có một số lỗi tiềm tàng, thứ tôi đã bỏ lỡ, v.v. Ngoài ra, nếu tôi có thể hỏi: vì tôi không hoàn toàn hiểu cách hoạt động của đồng thời Net::Curl::Multi, hãy cho tôi biết liệu tôi có nên gặp bất kỳ vấn đề nào với việc đặt lệnh UPDATE của MySQL (thông qua DBI) bên trong vòng lặp RESPONSE (ngoài tải máy chủ cao hơn rõ ràng) kịch bản để chạy với khoảng 50 đồng thời N::C::M công nhân, có thể nhiều hơn).

#!/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 là một thư viện khá tốt cực kỳ nhanh. Hơn nữa, nó có thể xử lý các yêu cầu song song quá! Tôi khuyên bạn nên sử dụng nó thay vì 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

này thực hiện chính xác những gì bạn muốn, một cách không đồng bộ, và nó làm điều đó bằng cách gói Net::Curl một cách an toàn:

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