Пишем простой асинхронный парсер

Многие разработчики типового ~~говно~~софта, ориентированного на работу с вебом, зачастую используют потоки для того, чтобы получить выигрыш в скорости. Данный подход, конечно, обладает своими плюсами, но все же не является оптимальным, например, с точки зрения потребляемых ресурсов системы (особенно когда речь идет о потребителях, любящих ставить сразу "тыщу потоков").
Альтернативным и общеизвестным способом ускорения работы софта является асинхронная модель, то есть модель, при которой все вызовы методов являются неблокирующими. В данной статье я рассмотрю простой пример, который будет использовать асинхронные веб-запросы.
В качестве примера будет написан парсер идентификаторов приложений с Android Market, который пригодится в готовящейся статье, посвященной добычи трафика с маркета. Для простоты будем использовать модуль AnyEvent, он упрощает реализацию асинхронной событийной модели. Итак, приступим.

Для начала прагмы, необходимые инклюды и переменные:

use strict;
use warnings;
use AnyEvent::HTTP;
use Fcntl qw/:flock/;

#Список категорий, которые скрипт будет обрабатывать
my @cat_list = qw/ARCADE BRAIN CARDS CASUAL GAME_WALLPAPER RACING SPORTS_GAMES GAME_WIDGETS/;
#Лимит страниц для каждой категории
my $page_limit = 15;
#Файл для сохранения результата
my $res_file = 'result1.txt';

#Autoflush
$| = 1;
#Для последующего определения времени работы скрипта
my $time = time;

use strict;

use warnings;

use AnyEvent::HTTP;

use Fcntl qw/:flock/;

#Список категорий, которые скрипт будет обрабатывать

my @cat_list = qw/ARCADE BRAIN CARDS CASUAL GAME_WALLPAPER RACING SPORTS_GAMES GAME_WIDGETS/;

#Лимит страниц для каждой категории

my $page_limit = 15;

#Файл для сохранения результата

my $res_file = 'result1.txt';

#Autoflush

$| = 1;

#Для последующего определения времени работы скрипта

my $time = time;

В целом код довольно небольшой, поэтому я не буду излишне фрагментировать его, а изложу основную часть одним куском и прокомментирую.

#Перебираем категории в массиве
for my $cat_name(@cat_list)
{
    #Создаем переменную, необходимую для последующей блокировки дальнейшего выполнения скрипта
    #(чтобы выполнять запросы пачками, по 15)
    #http://search.cpan.org/~mlehmann/AnyEvent-6.14/lib/AnyEvent.pm#CONDITION_VARIABLES
    my $cv = AnyEvent->condvar;
    #Ставим в очередь гет-запросы
    for(my ($i, $j) = (0, 0); $i < $page_limit; $i++)
    {
        print $cat_name, ' - ', $i, $/;
        #Определяем параметры гет-запроса
        http_get
        (
            'https://market.android.com/details?id=apps_topselling_paid&cat='.$cat_name.'&start='.($i * 25).'&num=25',
            headers => 
            {
                'User-Agent' => 'Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.2)'
            },
            timeout => 5,
            #Колбэк, который вызывается после выполнения запроса
            sub 
            {
                my ($data, $headers) = @_;
                #Если страница не существует, то android market возвращает 404
                #Проверяем этот момент
                if($headers->{Status} =~ /^200/)
                {
                    #Сохраняем в файл идентификаторы приложений
                    open F, '>>', $res_file or warn $!;
                    flock F, LOCK_EX;
                    
                    print F (join $/, $data =~ /data-docid="(.+?)"/g), $/;
                    
                    flock F, LOCK_UN;
                    close F;
                }
                #Если все страницы были обработаны, 
                if(++$j == $page_limit)
                {
                    #Снимаем блокировку, чтобы поставить в очередь следующую пачку запросов
                    $cv->send;
                }
            }
        );
    }
    #Блокируем переменную, ждем пока все запросы будут обработаны
    #(как-бы ждем "сигнала", когда будет вызван метод ->send)
    $cv->recv;
}
#Выводим время работы скрипта
print 'Time elapsed: ', time - $time, $/;

#Перебираем категории в массиве

for my $cat_name(@cat_list)

{

#Создаем переменную, необходимую для последующей блокировки дальнейшего выполнения скрипта

#(чтобы выполнять запросы пачками, по 15)

#http://search.cpan.org/~mlehmann/AnyEvent-6.14/lib/AnyEvent.pm#CONDITION_VARIABLES

my $cv = AnyEvent->condvar;

#Ставим в очередь гет-запросы

for(my ($i, $j) = (0, 0); $i < $page_limit; $i++)

{

print $cat_name, ' - ', $i, $/;

#Определяем параметры гет-запроса

http_get

(

'https://market.android.com/details?id=apps_topselling_paid&cat='.$cat_name.'&start='.($i * 25).'&num=25',

headers =>

{

'User-Agent' => 'Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.2)'

timeout => 5,

#Колбэк, который вызывается после выполнения запроса

sub

{

my ($data, $headers) = @_;

#Если страница не существует, то android market возвращает 404

#Проверяем этот момент

if($headers->{Status} =~ /^200/)

{

#Сохраняем в файл идентификаторы приложений

open F, '>>', $res_file or warn $!;

flock F, LOCK_EX;

print F (join $/, $data =~ /data-docid="(.+?)"/g), $/;

flock F, LOCK_UN;

close F;

}

#Если все страницы были обработаны,

if(++$j == $page_limit)

{

#Снимаем блокировку, чтобы поставить в очередь следующую пачку запросов

$cv->send;

}

);

}

#Блокируем переменную, ждем пока все запросы будут обработаны

#(как-бы ждем "сигнала", когда будет вызван метод ->send)

$cv->recv;

}

#Выводим время работы скрипта

print 'Time elapsed: ', time - $time, $/;

Теперь сравним вышеприведенный скрипт со скриптом, который выполняет ту же самую работу, но опирается на потоки. Вот код этого скрипта:

use strict;
use warnings;
use threads;
use threads::shared;
use IO::Socket::SSL;
use Fcntl qw/:flock/;


my $thr_cnt = 15;
my @cat_list : shared = qw/ARCADE BRAIN CARDS CASUAL GAME_WALLPAPER RACING SPORTS_GAMES GAME_WIDGETS/;
my $page_limit = 15;
my $res_file = 'result2.txt';

$| = 1;
my @trl = ();
my $w_lock : shared;

my $time = time;

$trl[$_] = threads->create(\&main) for 0..$thr_cnt - 1;
$_->join for @trl;

print 'Time elapsed: ', time - $time, $/;

sub main
{
    while(1)
    {
        my $cat_name = shift @cat_list or last;
        
        for(my $i = 0; $i < $page_limit; $i++)
        {
            print $cat_name, ' - ', $i, $/;
            
            my $socket = IO::Socket::SSL->new
            (
                PeerAddr => 'market.android.com',
                PeerPort => 443,
                PeerProto => 'tcp', 
                TimeOut => 5
            );
            
            if($socket)
            {
                my $req =
                "GET /details?id=apps_topselling_paid&cat=$cat_name&start=".($i * 25)."&num=25 HTTP/1.0\r\n".
                "Host: market.android.com\r\n".
                "User-Agent: Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.2)\r\n".
                "Connection: Close\r\n\r\n";
                
                print $socket $req;
                
                my $data;
                while(read $socket, my $buffer, 128)
                {
                    $data .= $buffer;
                }
                
                close $socket;
                
                if($data =~ /200 OK/)
                {
                    lock $w_lock;
                    open F, '>>', $res_file or warn $!;
                    flock F, LOCK_EX;
                    
                    print F (join $/, $data =~ /data-docid="(.+?)"/g), $/;
                    
                    flock F, LOCK_UN;
                    close F;
                }
            }
        }
    }
}

use strict;

use warnings;

use threads;

use threads::shared;

use IO::Socket::SSL;

use Fcntl qw/:flock/;

my $thr_cnt = 15;

my @cat_list : shared = qw/ARCADE BRAIN CARDS CASUAL GAME_WALLPAPER RACING SPORTS_GAMES GAME_WIDGETS/;

my $page_limit = 15;

my $res_file = 'result2.txt';

$| = 1;

my @trl = ();

my $w_lock : shared;

my $time = time;

$trl[$_] = threads->create(\&main) for 0..$thr_cnt - 1;

$_->join for @trl;

print 'Time elapsed: ', time - $time, $/;

sub main

{

while(1)

{

my $cat_name = shift @cat_list or last;

for(my $i = 0; $i < $page_limit; $i++)

{

print $cat_name, ' - ', $i, $/;

my $socket = IO::Socket::SSL->new

(

PeerAddr => 'market.android.com',

PeerPort => 443,

PeerProto => 'tcp',

TimeOut => 5

);

if($socket)

{

my $req =

"GET /details?id=apps_topselling_paid&cat=$cat_name&start=".($i * 25)."&num=25 HTTP/1.0\r\n".

"Host: market.android.com\r\n".

"User-Agent: Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.2)\r\n".

"Connection: Close\r\n\r\n";

print $socket $req;

my $data;

while(read $socket, my $buffer, 128)

{

$data .= $buffer;

}

close $socket;

if($data =~ /200 OK/)

{

lock $w_lock;

open F, '>>', $res_file or warn $!;

flock F, LOCK_EX;

print F (join $/, $data =~ /data-docid="(.+?)"/g), $/;

flock F, LOCK_UN;

close F;

}

Скорость работы скриптов примерно одинакова, и там и там идет одновременное выполнение 15 веб-запросов, однако, если обратить внимание на потребляемую память и нагрузку на процессор, то мы увидим явное преимущество у первого скрипта. У меня количество потребляемой памяти для асинхронного варианта составляло ~ 12 мегабайт, а в случае с потоками оно увеличилось до ~ 53 мегабайт (хотя прожорливость perl + pthreads под win* - известная печалька).
Но даже если абстрагироваться от языка программирования и использовать WinAPI для работы с потоками, то можно увидеть, что при большом количестве потоков (конечно, ведь чем больше потоков, тем круче и быстрее (с) дефолт логика) нагрузка на процессор вырастет несоизмеримо по сравнению с асинхронной моделью.

В общем, асинхронная модель зачастую является более предпочтительной, но иногда требует нетривиального планирования алгоритма работы программы, например, когда есть необходимость совершать множество зависящих друг от друга запросов и менять логику поведения в зависимости от результатов, полученных на предыдущих стадиях.

Скрипты из статьи: скачать

Пишем простой асинхронный парсер: 17 комментариев

flisk:

24 февраля, 2012 в 01:23

Спасибо за интересный пример, давно хотел попробовать поработать с AnyEvent.

Ответить
Рита:

24 февраля, 2012 в 15:11

>конечно, ведь чем больше потоков, тем круче и быстрее (с) дефолт логика
Дефолт сарказм обладателя полуядерного ЦП

Ответить
1. dx:
  
  24 февраля, 2012 в 15:41
  
  Риточка, Вы не правы. Если Вы поставите 1000 потоков на четырехядерной машине, Вы уже увидите, что приличная часть процессорного времени расходуется на переключение их контекстов. Если речь идет о Windows и перле, то вы и 100 потоков с трудом поставите, так как у вас оперативной памяти несколько гигабайт отожрется при этом. А если Вы поддерживаете мысль "1000 потоков - это охуенно", то советуем Вам немного пересмотреть свою логику. Хотя, вероятно, Вы просто can't into asynchronous.
  
  Ответить
  1. Рита:
    
    24 февраля, 2012 в 18:21
    
    Вот она система моей мечты, Windows, конечно! Весь софт серьезные дядьки - спамеры веба пишут под неё, ага.
    
    Ответить
    1. Дядька-спамер:
      
      24 февраля, 2012 в 19:19
      
      Серьезные дядьки-спамеры по социалочкам голосуют за Windows из-за наличия кучи дешевых дедиков! Windows - выбор профессионала.
      
      Ответить
2. mr.The:
  
  24 февраля, 2012 в 15:53
  
  >Дефолт сарказм обладателя полуядерного ЦП
  дефолт сарказм владельца мощного пк купленного за мамины деньги.
  
  Ответить
Василий:

24 февраля, 2012 в 16:18

Что-то я не понял, зачем вам тут (f)lock понадобился. Приложение однопоточное, в каждый момент времени у вас может выполняться только один колбэк. Вопрос: зачем блокировка (еще и двойная)?

Ответить
1. Kaimi:
  
  24 февраля, 2012 в 17:10
  
  ctrl+c - ctrl+v
  
  Ответить
zloid:

26 февраля, 2012 в 10:19

Добыча трафика с маркета, это что такое? Переманивание посетителей с маркета на какой-то свой сайт? Или раскрутка своего приложения на маркете?

Ответить
1. Kaimi:
  
  26 февраля, 2012 в 14:28
  
  1ое
  
  Ответить
БыдлоКодер:

1 марта, 2012 в 04:53

Kaimi, ты так ненавидишь быдло/говно кодеров, но почему ты сам сам активно занимаешься быдло/говно кодингом за деньги? Или это чисто прагматическая ненависть к конкурентам?

Ответить
1. zloid:
  
  1 марта, 2012 в 21:33
  
  Уходи, а то я тебя по айпи вычислю.
  
  Ответить
  1. БыдлоКодер:
    
    1 марта, 2012 в 23:43
    
    Пощади, о ужасный Ктулху.
    
    Ответить
Уведомление: Добываем трафик c помощью Android Market
Kairos:

13 мая, 2012 в 11:34

А в C# теперь есть async/await, благодаря чему IO completion ports использовать так же удобно, как и потоки.

Ответить
user:

9 июня, 2013 в 13:48

Подскажите как можно использовать AnyEvent::Socket? На CPAN этот модуль есть, а Perl Packet Manager его не отображает для установки.

Ответить
1. Kaimi:
  
  9 июня, 2013 в 14:24
  
  http://www.activestate.com/blog/2010/10/how-install-cpan-modules-activeperl
  
  Ответить