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

iamlimeng

我经常需要提取大量的（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次；

1. #!/usr/bin/perl
   
2. 
   
3. use strict;
   
4. use warnings;
   
5. use LWP::UserAgent;
   
6. use Benchmark;
   
7. my $TT0 = new Benchmark;
   
8. 
   
9. my $url = "http://www.baidu.com";
   
10. my $request_times = 1000;
    
11. 
    
12. print "\n Now begin testing ... \n";
    
13. my $lwp = new LWP::UserAgent(agent => 'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727; CIBA)');
    
14. 
    
15. for(1..$request_times) {
    
16.         my $request = HTTP::Request->new(GET=>$url);
    
17.         $request->header(Accept=>'text/html');
    
18.         my $response = $lwp->request($request);
    
19.          if ($response->is_success) {
    
20.                  print " $_\tOK!\n";
    
21.          }
    
22.          else {
    
23.                  print " $_\tFaild!\n";
    
24.                  redo;
    
25.          }
    
26. }
    
27. my $TT1 = new Benchmark;
    
28. my $td = Benchmark::timediff($TT1, $TT0);
    
29. $td = Benchmark::timestr($td);
    
30. my ($sec) = ($td =~ /(\d+).*/);
    
31. my $speed = sprintf("%0.1f",$request_times/$sec);
    
32. print "\n Time expend: $td\n Average Speed: $speed Times Per Second\n\n Press Enter to close me ... \7";
    
33. 
    
34. <STDIN>;

复制代码

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

1. #!/usr/bin/perl
   
2. 
   
3. use strict;
   
4. use warnings;
   
5. use LWP::UserAgent;
   
6. use LWP::ConnCache;
   
7. use Benchmark;
   
8. my $TT0 = new Benchmark;
   
9. 
   
10. my $url = "http://www.baidu.com";
    
11. my $request_times = 1000;
    
12. 
    
13. print "\n Now begin testing ... \n";
    
14. my $lwp = new LWP::UserAgent(agent => 'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727; CIBA)');
    
15. my $conncache = new LWP::ConnCache;
    
16. $lwp->conn_cache($conncache);
    
17. 
    
18. for(1..$request_times) {
    
19.         my $request = HTTP::Request->new(GET=>$url);
    
20.         $request->header(Accept=>'text/html');
    
21.         my $response = $lwp->request($request);
    
22.          if ($response->is_success) {
    
23.                  print " $_\tOK!\n";
    
24.          }
    
25.          else {
    
26.                  print " $_\tFaild!\n";
    
27.                  redo;
    
28.          }
    
29. }
    
30. my $TT1 = new Benchmark;
    
31. my $td = Benchmark::timediff($TT1, $TT0);
    
32. $td = Benchmark::timestr($td);
    
33. my ($sec) = ($td =~ /(\d+).*/);
    
34. my $speed = sprintf("%0.1f",$request_times/$sec);
    
35. print "\n Time expend: $td\n Average Speed: $speed Times Per Second\n\n Press Enter to close me ... \7";
    
36. 
    
37. <STDIN>;

复制代码

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

1. #!/usr/bin/perl
   
2. 
   
3. use strict;
   
4. use warnings;
   
5. use threads;
   
6. use threads::shared;
   
7. use Thread::Queue;
   
8. use LWP::UserAgent;
   
9. use LWP::ConnCache;
   
10. use Benchmark;
    
11. my $TT0 = new Benchmark;
    
12. 
    
13. my $url = "http://www.baidu.com";
    
14. my $request_times = 1000;
    
15. 
    
16. print "\n Now begin testing ... \n";
    
17. my $lwp = new LWP::UserAgent(agent => 'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727; CIBA)');
    
18. my $conncache = new LWP::ConnCache;
    
19. $lwp->conn_cache($conncache);
    
20. 
    
21. my $data_queue = new Thread::Queue;
    
22. my $result_queue = new Thread::Queue;
    
23. my $processing_count :shared = 0;
    
24. my $MAX_THREADS = 10;
    
25. my $num = 1;
    
26. 
    
27. for (my $n = 0; $n < $MAX_THREADS; $n++)
    
28. {
    
29.         threads->create(\&thread_io);
    
30. }
    
31. 
    
32. foreach my $data(1..$request_times)
    
33. {
    
34.         if ($data_queue ->pending() > $MAX_THREADS * 2)
    
35.         {
    
36.                 select(undef, undef, undef, 0.02);
    
37.                 redo;
    
38.         }
    
39. 
    
40.         $data_queue->enqueue($data);
    
41.         if ($result_queue->pending() > 0)
    
42.         {
    
43.                 while (my $result = $result_queue->dequeue_nb())
    
44.                 {
    
45.                         if($result) { print " $num\tOK!\n"; }
    
46.                         else { print " $num\tFailed!\n"; }
    
47.                          $num++;
    
48.                 }
    
49.         }
    
50. }
    
51. 
    
52. while ($processing_count > 0 or $data_queue->pending() > 0 or $result_queue->pending() > 0)
    
53. {
    
54.          select(undef, undef, undef, 0.02);
    
55.         while (my $result = $result_queue->dequeue_nb())
    
56.         {
    
57.                         if($result) { print " $num\tOK!\n"; }
    
58.                         else { print " $num\tFailed!\n"; }
    
59.                          $num++;
    
60.         }
    
61. }
    
62. 
    
63. foreach my $thread (threads->list(threads::all))
    
64. {
    
65.         $thread->detach();
    
66. }
    
67. 
    
68. my $TT1 = new Benchmark;
    
69. my $td = Benchmark::timediff($TT1, $TT0);
    
70. $td = Benchmark::timestr($td);
    
71. my ($sec) = ($td =~ /(\d+).*/);
    
72. my $speed = sprintf("%0.1f",$request_times/$sec);
    
73. print "\n Time expend: $td\n Average Speed: $speed Times Per Second\n\n Press Enter to close me ... \7";
    
74. 
    
75. <STDIN>;
    
76. 
    
77. ##########################################################################################
    
78. 
    
79. sub thread_io()
    
80. {
    
81.         while (my $data = $data_queue->dequeue())
    
82.         {
    
83.                 {
    
84.                         lock $processing_count;
    
85.                         ++$processing_count;
    
86.                 }
    
87. 
    
88.                 my $result = get_html($data);
    
89.                 $result_queue->enqueue($result);
    
90. 
    
91.                 {
    
92.                         lock $processing_count;
    
93.                         --$processing_count;
    
94.                 }
    
95.         }
    
96. }
    
97. 
    
98. sub get_html {
    
99.         my $no = shift;
    
100.         my $request = HTTP::Request->new(GET=>$url);
     
101.         $request->header(Accept=>'text/html');
     
102.         my $response = $lwp->request($request);
     
103.          if ($response->is_success) {
     
104.                  return(1);
     
105.          }
     
106.          else {
     
107.                 $data_queue->enqueue($no);   #请求失败的任务，可以在此重新加入队列
     
108.                  return(0);
     
109.          }
     
110. }

复制代码

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

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

在此特别感谢莫言。

yafeiie

不错，收藏并研究中。。。

兰花仙子

但仙子给的多线程模型很难控制，下载过程中经常发呆很久

那只是个模型而已的。。
产品级线程/进程应用，不是那几行代码就可以的。

黑色阳光_cu

CORO版本的线程池

1. 
   
2. #!/bin/env perl
   
3. 
   
4. use strict;
   
5. use warnings;
   
6. use Benchmark;
   
7. use Coro;
   
8. use Coro::Timer;
   
9. use Coro::LWP;
   
10. use LWP::UserAgent;
    
11. use LWP::ConnCache;
    
12. 
    
13. my $MAX_THREADS = 10;
    
14. my @data_queue;
    
15. my @result_queue;
    
16. my $processing_count = 0;
    
17. 
    
18. my $TT0 = new Benchmark;
    
19. 
    
20. foreach (1 .. 10)
    
21. {
    
22.         async(\&thread_io);
    
23. }
    
24. 
    
25. foreach (1 .. 1000)
    
26. {
    
27.         if ($#data_queue > $MAX_THREADS * 2)
    
28.         {
    
29.                 Coro::Timer::sleep(0.02);
    
30.                 redo;
    
31.         }
    
32. 
    
33.         push(@data_queue, $_);
    
34.         if ($#result_queue > -1)
    
35.         {
    
36.                 warn $#result_queue;
    
37.                 while ($#result_queue > -1)
    
38.                 {
    
39.                         my $result = shift(@result_queue);
    
40.                         print "$result\n";
    
41.                 }
    
42.         }
    
43. }
    
44. 
    
45. while ($processing_count > 0 or $#data_queue > -1 or $#result_queue > -1)
    
46. {
    
47.         while ($#result_queue > -1)
    
48.         {
    
49.                 my $result = shift(@result_queue);
    
50.                 print "$result\n";
    
51.         }
    
52.        
    
53.         Coro::Timer::sleep(0.02);
    
54. }
    
55. 
    
56. my $TT1 = new Benchmark;
    
57. my $td = Benchmark::timediff($TT1, $TT0);
    
58. $td = Benchmark::timestr($td);
    
59. print "\n Time expend: $td\n\n Press Enter to close me ... \7";
    
60. <STDIN>;
    
61. 
    
62. ##########################################################################
    
63. 
    
64. sub thread_io()
    
65. {
    
66.         my $lwp = LWP::UserAgent->new();
    
67.         my $conncache = new LWP::ConnCache;
    
68.         $lwp->conn_cache($conncache);
    
69. 
    
70.         while (1)
    
71.         {
    
72.                 if ($#data_queue == -1)
    
73.                 {
    
74.                         Coro::Timer::sleep(1);
    
75.                         next;
    
76.                 }
    
77. 
    
78.                 my $data = shift(@data_queue);
    
79.                 ++$processing_count;
    
80.                 my $r = $lwp->get("http://www.baidu.com/");
    
81.                 my $status = $r->status_line();
    
82.                 push(@result_queue, "$Coro::current $data $status");
    
83.                 --$processing_count;
    
84.         }
    
85. }
    
86. 
    

复制代码

yz86yz

恩 学习下

xuri0419

Perl China官方交流群59785891

此莫言是这个群的莫言吗？

hitsubunnu

Coro直接带 Coro::Semaphore; 线程池 不用自己写

1. use strict;
   
2. use warnings;
   
3. use Smart::Comments;
   
4. 
   
5. use Coro;
   
6. use Coro::LWP;
   
7. use Coro::Semaphore;
   
8. use LWP::UserAgent;
   
9. use LWP::ConnCache;
   
10. use constant LIMIT => 200;
    
11. 
    
12. use Benchmark;
    
13. my $TT0 = new Benchmark;
    
14. my $request_times = 1000;
    
15. print "\n Now begin testing ... \n";
    
16. 
    
17. my $outpath = './data';
    
18. my $url = 'http://www.baidu.com';
    
19. 
    
20. my $semaphore = Coro::Semaphore->new(LIMIT);
    
21. my (@coros,@error);
    
22. 
    
23. my $ua    = new LWP::UserAgent;
    
24. my $conncache = new LWP::ConnCache;
    
25. $ua->conn_cache($conncache);
    
26. 
    
27. for my $num (0..999) {
    
28.         my $outfile = $outpath.'/'.$num.'.html';
    
29.         my $outfile_tmp = $outfile.'.tmp';
    
30.     push @coros, async {
    
31.                my $guard = $semaphore->guard;
    
32.        
    
33.                 my $req = HTTP::Request->new(GET => $url);
    
34.                 my $res = $ua->request($req);
    
35.                                                                                                        
    
36.                    if($res->is_success){
    
37.                            my $html = $res -> content;
    
38.                            open(FF,">",$outfile_tmp);
    
39.                            print FF $html;
    
40.                            close(FF);
    
41.                           
    
42.                            if(-e $outfile_tmp){
    
43.                                    unlink $outfile;
    
44.                                    rename $outfile_tmp,$outfile;
    
45.                            }
    
46.                            print $num,'ok',$/;
    
47.                    }else{
    
48.                            push @error,$url;
    
49.                    }
    
50.         };
    
51. }
    
52.        
    
53. $_ -> join for @coros;
    
54.        
    
55. my $TT1 = new Benchmark;
    
56. my $td = Benchmark::timediff($TT1, $TT0);
    
57. $td = Benchmark::timestr($td);
    
58. my ($sec) = ($td =~ /(\d+).*/);
    
59. my $speed = sprintf("%0.1f",$request_times/$sec);
    
60. print "\n Time expend: $td\n Average Speed: $speed Times Per Second\n\n Press Enter to close me ... \7";
    
61. 
    

复制代码

iamlimeng

Perl China官方交流群59785891

此莫言是这个群的莫言吗？
xuri0419 发表于 2010-06-17 09:51

    正是！

黑色阳光_cu

Coro直接带 Coro::Semaphore; 线程池 不用自己写
hitsubunnu 发表于 2010-06-17 11:25

没看明白~~~

lucash

回复 4# 黑色阳光_cu


   
哈哈!.