保留] 用多线程和持续连接实现高速WEB请求

时间：2010-09-15 来源：snowtty

我经常需要提取大量的（1500页以上）网页数据，曾尝试过很多方法，虽然都能实现，但效率都不是太高。

刚开始用LWP::Simple（get）按顺序边下载边提取，这种方法很容易控制，也很可靠，下载中途中断了可以通过检查数据的完整性断点续传，下载的网页数据并不存入本地硬盘，仅存储提取后的少量数据，硬盘操作少，但下载效率很低；

为了加快下载速度，考虑用第三方下载软件先下载网页数据，再用程序从已下载的网页中提取数据。所以开始使用Teleport下载网页数据，Teleport支持多线程下载，速度提高了至少3－5倍。但用Teleport下载大量网页的时候，也会有下载失败的情况，程序本身并不自动检测并重新下载，通常需要手工重复下载一次以检验数据的完整性，这需要额外消耗一些时间。这种方法效率较高，也很稳定，我使用了很长时间；

后来在CU论坛看到仙子发的多线程模型，就将自己的代码改造成多线程下载，效率比Teleport快了不少，但仙子给的多线程模型很难控制，下载过程中经常发呆很久，下载失败率很高（10%左右），下载失败的任务无法再继续通过多线程下载（我没找到方法），不能即时显示下载进度，在下载后期经常假死，处于无限期等待状态，程序不再继续运行，不得不手工关闭。仙子发的多进程模型也尝试过，多线程中的问题，多进程同样存在，而且资源消耗非常大。所以不得不放弃，曾对PERL的多线程和多进程不抱希望；

再后来，Perl China官方QQ群群主莫言给了个终极解决方案，使用LWP::ConnCache建立持续连接，并结合多线程（线程池方式）实现高速WEB数据请求。这种方案非常高效，下载速度是Teleport的3－5倍，而且易于控制，能即时显示下载进度。

现共享给大家，以求共同进步。

（以下代码以请求1000次百度主页为例，下载测试环境为：XP，ActivePerl 5.10.1007，1M电信宽带，测试数据仅供参考）

1.顺序请求，不使用持续连接，程序每请求一次需要与服务器新建一次连接，这会耗费大量时间，平均下载速度约为每秒0.7次；

#!/usr/bin/perl
use strict;
use warnings;
use LWP::UserAgent;
use Benchmark;
my $TT0 = new Benchmark;
my $url = "http://www.baidu.com";
my $request_times = 1000;
print "\n Now begin testing ... \n";
my $lwp = new LWP::UserAgent(agent => 'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727; CIBA)');
for(1..$request_times) {
my $request = HTTP::Request->new(GET=>$url);
$request->header(Accept=>'text/html');
my $response = $lwp->request($request);
if ($response->is_success) {
print " $_\tOK!\n";
}
else {
print " $_\tFaild!\n";
redo;
}
}
my $TT1 = new Benchmark;
my $td = Benchmark::timediff($TT1, $TT0);
$td = Benchmark::timestr($td);
my ($sec) = ($td =~ /(\d+).*/);
my $speed = sprintf("%0.1f",$request_times/$sec);
print "\n Time expend: $td\n Average Speed: $speed Times Per Second\n\n Press Enter to close me ... \7";
<STDIN>;

复制代码 2.顺序请求，使用持续连接，向同一服务器多次发送请求仅需建立一次连接，平均下载速度约为每秒7.2次，下载效率有明显提高；

#!/usr/bin/perl
use strict;
use warnings;
use LWP::UserAgent;
use LWP::ConnCache;
use Benchmark;
my $TT0 = new Benchmark;
my $url = "http://www.baidu.com";
my $request_times = 1000;
print "\n Now begin testing ... \n";
my $lwp = new LWP::UserAgent(agent => 'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727; CIBA)');
my $conncache = new LWP::ConnCache;
$lwp->conn_cache($conncache);
for(1..$request_times) {
my $request = HTTP::Request->new(GET=>$url);
$request->header(Accept=>'text/html');
my $response = $lwp->request($request);
if ($response->is_success) {
print " $_\tOK!\n";
}
else {
print " $_\tFaild!\n";
redo;
}
}
my $TT1 = new Benchmark;
my $td = Benchmark::timediff($TT1, $TT0);
$td = Benchmark::timestr($td);
my ($sec) = ($td =~ /(\d+).*/);
my $speed = sprintf("%0.1f",$request_times/$sec);
print "\n Time expend: $td\n Average Speed: $speed Times Per Second\n\n Press Enter to close me ... \7";
<STDIN>;

复制代码 3.使用多线程，并结合持续连接。平均下载速度约为每秒17次，是令人兴奋的质的飞跃。

#!/usr/bin/perl
use strict;
use warnings;
use threads;
use threads::shared;
use Thread::Queue;
use LWP::UserAgent;
use LWP::ConnCache;
use Benchmark;
my $TT0 = new Benchmark;
my $url = "http://www.baidu.com";
my $request_times = 1000;
print "\n Now begin testing ... \n";
my $lwp = new LWP::UserAgent(agent => 'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727; CIBA)');
my $conncache = new LWP::ConnCache;
$lwp->conn_cache($conncache);
my $data_queue = new Thread::Queue;
my $result_queue = new Thread::Queue;
my $processing_count :shared = 0;
my $MAX_THREADS = 10;
my $num = 1;
for (my $n = 0; $n < $MAX_THREADS; $n++)
{
threads->create(\&thread_io);
}
foreach my $data(1..$request_times)
{
if ($data_queue ->pending() > $MAX_THREADS * 2)
{
select(undef, undef, undef, 0.02);
redo;
}
$data_queue->enqueue($data);
if ($result_queue->pending() > 0)
{
while (my $result = $result_queue->dequeue_nb())
{
if($result) { print " $num\tOK!\n"; }
else { print " $num\tFailed!\n"; }
$num++;
}
}
}
while ($processing_count > 0 or $data_queue->pending() > 0 or $result_queue->pending() > 0)
{
select(undef, undef, undef, 0.02);
while (my $result = $result_queue->dequeue_nb())
{
if($result) { print " $num\tOK!\n"; }
else { print " $num\tFailed!\n"; }
$num++;
}
}
foreach my $thread (threads->list(threads::all))
{
$thread->detach();
}
my $TT1 = new Benchmark;
my $td = Benchmark::timediff($TT1, $TT0);
$td = Benchmark::timestr($td);
my ($sec) = ($td =~ /(\d+).*/);
my $speed = sprintf("%0.1f",$request_times/$sec);
print "\n Time expend: $td\n Average Speed: $speed Times Per Second\n\n Press Enter to close me ... \7";
<STDIN>;
##########################################################################################
sub thread_io()
{
while (my $data = $data_queue->dequeue())
{
{
lock $processing_count;
++$processing_count;
}
my $result = get_html($data);
$result_queue->enqueue($result);
{
lock $processing_count;
--$processing_count;
}
}
}
sub get_html {
my $no = shift;
my $request = HTTP::Request->new(GET=>$url);
$request->header(Accept=>'text/html');
my $response = $lwp->request($request);
if ($response->is_success) {
return(1);
}
else {
$data_queue->enqueue($no); #请求失败的任务，可以在此重新加入队列
return(0);
}
}

复制代码该多线程模型为莫言原创，采用线程池方式，建2个队列，一个负责向线程队列添加任务，另一个负责管理任务的处理结果。请求失败的任务，可以重新加入队列，以保证每个请求的有效性。该模型易于控制，可靠性高，能即时显示任务的处理进度，而且效率高。

注意：LWP::ConnCache不支持LWP::Simple，支持LWP::UserAgent。

在此特别感谢莫言。

#!/usr/bin/perl
use strict;
use warnings;
use LWP::UserAgent;
use Benchmark;
my $TT0 = new Benchmark;
my $url = "http://www.baidu.com";
my $request_times = 1000;
print "\n Now begin testing ... \n";
my $lwp = new LWP::UserAgent(agent => 'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727; CIBA)');
for(1..$request_times) {
my $request = HTTP::Request->new(GET=>$url);
$request->header(Accept=>'text/html');
my $response = $lwp->request($request);
if ($response->is_success) {
print " $_\tOK!\n";
}
else {
print " $_\tFaild!\n";
redo;
}
}
my $TT1 = new Benchmark;
my $td = Benchmark::timediff($TT1, $TT0);
$td = Benchmark::timestr($td);
my ($sec) = ($td =~ /(\d+).*/);
my $speed = sprintf("%0.1f",$request_times/$sec);
print "\n Time expend: $td\n Average Speed: $speed Times Per Second\n\n Press Enter to close me ... \7";
<STDIN>;

复制代码 2.顺序请求，使用持续连接，向同一服务器多次发送请求仅需建立一次连接，平均下载速度约为每秒7.2次，下载效率有明显提高；

#!/usr/bin/perl
use strict;
use warnings;
use LWP::UserAgent;
use LWP::ConnCache;
use Benchmark;
my $TT0 = new Benchmark;
my $url = "http://www.baidu.com";
my $request_times = 1000;
print "\n Now begin testing ... \n";
my $lwp = new LWP::UserAgent(agent => 'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727; CIBA)');
my $conncache = new LWP::ConnCache;
$lwp->conn_cache($conncache);
for(1..$request_times) {
my $request = HTTP::Request->new(GET=>$url);
$request->header(Accept=>'text/html');
my $response = $lwp->request($request);
if ($response->is_success) {
print " $_\tOK!\n";
}
else {
print " $_\tFaild!\n";
redo;
}
}
my $TT1 = new Benchmark;
my $td = Benchmark::timediff($TT1, $TT0);
$td = Benchmark::timestr($td);
my ($sec) = ($td =~ /(\d+).*/);
my $speed = sprintf("%0.1f",$request_times/$sec);
print "\n Time expend: $td\n Average Speed: $speed Times Per Second\n\n Press Enter to close me ... \7";
<STDIN>;

复制代码 3.使用多线程，并结合持续连接。平均下载速度约为每秒17次，是令人兴奋的质的飞跃。

#!/usr/bin/perl
use strict;
use warnings;
use threads;
use threads::shared;
use Thread::Queue;
use LWP::UserAgent;
use LWP::ConnCache;
use Benchmark;
my $TT0 = new Benchmark;
my $url = "http://www.baidu.com";
my $request_times = 1000;
print "\n Now begin testing ... \n";
my $lwp = new LWP::UserAgent(agent => 'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727; CIBA)');
my $conncache = new LWP::ConnCache;
$lwp->conn_cache($conncache);
my $data_queue = new Thread::Queue;
my $result_queue = new Thread::Queue;
my $processing_count :shared = 0;
my $MAX_THREADS = 10;
my $num = 1;
for (my $n = 0; $n < $MAX_THREADS; $n++)
{
threads->create(\&thread_io);
}
foreach my $data(1..$request_times)
{
if ($data_queue ->pending() > $MAX_THREADS * 2)
{
select(undef, undef, undef, 0.02);
redo;
}
$data_queue->enqueue($data);
if ($result_queue->pending() > 0)
{
while (my $result = $result_queue->dequeue_nb())
{
if($result) { print " $num\tOK!\n"; }
else { print " $num\tFailed!\n"; }
$num++;
}
}
}
while ($processing_count > 0 or $data_queue->pending() > 0 or $result_queue->pending() > 0)
{
select(undef, undef, undef, 0.02);
while (my $result = $result_queue->dequeue_nb())
{
if($result) { print " $num\tOK!\n"; }
else { print " $num\tFailed!\n"; }
$num++;
}
}
foreach my $thread (threads->list(threads::all))
{
$thread->detach();
}
my $TT1 = new Benchmark;
my $td = Benchmark::timediff($TT1, $TT0);
$td = Benchmark::timestr($td);
my ($sec) = ($td =~ /(\d+).*/);
my $speed = sprintf("%0.1f",$request_times/$sec);
print "\n Time expend: $td\n Average Speed: $speed Times Per Second\n\n Press Enter to close me ... \7";
<STDIN>;
##########################################################################################
sub thread_io()
{
while (my $data = $data_queue->dequeue())
{
{
lock $processing_count;
++$processing_count;
}
my $result = get_html($data);
$result_queue->enqueue($result);
{
lock $processing_count;
--$processing_count;
}
}
}
sub get_html {
my $no = shift;
my $request = HTTP::Request->new(GET=>$url);
$request->header(Accept=>'text/html');
my $response = $lwp->request($request);
if ($response->is_success) {
return(1);
}
else {
$data_queue->enqueue($no); #请求失败的任务，可以在此重新加入队列
return(0);
}
}