Python はマルチスレッドを使用して Web ページ情報をクロールします

この記事では、Python のマルチスレッド Web ページ クローリング機能を主に紹介し、具体的な例に基づいて、Python マルチスレッド プログラミングの関連操作テクニックと注意事項を詳細に分析します。また、その実装方法を提供するデモ例も付属しています。マルチスレッド Web ページ クローリング、必要な友達は参考にしてください

この記事では、Python のマルチスレッド Web ページ クローリング機能の実装例について説明します。参考のために皆さんと共有してください。詳細は次のとおりです:

最近、私はウェブクローラーに関連したことをしています。 私はオープンソース C++ で書かれた larbin クローラーを調べ、設計上のアイデアといくつかの主要なテクノロジーの実装を注意深く読みました。

1. Larbin の URL 再利用は非常に効率的なブルーム フィルター アルゴリズムです。
2. DNS 処理では、一部がメモリにキャッシュされ、一部がファイルに書き込まれます。輸入戦略。
4. Larbin はファイル関連の操作に関して多くの作業を行っています。
5. larbin には接続プールがあり、HTTP プロトコルの GET メソッドをターゲット サイトに送信し、コンテンツを取得します。その後、ヘッダーなどを解析します。
6. 非常に効率的なポーリング方式による I/O 多重化に多数の記述子が使用されます。 7. Larbin は非常に構成可能です。 8. によって使用される多数のデータ構造。作者は下から自分で書いていて基本的には役に立たないSTL
…

なども今後時間があるときにまとめて記事にします。

過去 2 日間、私は Python でマルチスレッドのページ ダウンロード プログラムを作成しました。I/O 集中型のアプリケーションの場合、マルチスレッドは明らかに良い解決策です。先ほど書いたスレッドプールも使えます。実際、Python を使用してページをクロールするのは非常に簡単で、非常に使いやすく、基本的に 2 ～ 3 行のコードで実行できる urllib2 モジュールがあります。サードパーティのモジュールを使用すると問題を非常に便利に解決できますが、主要なアルゴリズムは自分自身ではなく他人によって実装されるため、詳細を理解できないだけです。 テクノロジーの専門家として、他の人が作成したモジュールや API を単に使用することはできません。それを自分で実装して、さらに学ぶ必要があります。

ソケットから開始することにしました。これは、GET プロトコルをカプセル化し、ヘッダーを解析することもできます。DNS キャッシュなどの DNS 解析プロセスも個別に処理できるため、自分で作成すると、より制御しやすくなります。拡張に役立ちます。タイムアウト処理には、グローバル 5 秒のタイムアウト処理を使用します。再配置 (301 または 302) 処理では、最大再配置は 3 回です。これは、以前のテスト プロセス中に、多くのサイトの再配置が自分自身にリダイレクトされることが判明したためです。無限ループになるので上限が設定されています。具体的な原理は比較的単純なので、コードを見てください。

書き終わった後、urllib2とパフォーマンスを比較してみたところ、自分で書いた方が効率が高く、理由はわかりませんが、urllib2の方がエラー率が若干高かったです。インターネット上の一部の人は、urllib2 にはマルチスレッド コンテキストでいくつかの小さな問題があると言っていますが、詳細については特に明確ではありません。

最初にコードを投稿します:

fetchPage.py

HttpプロトコルのGetメソッドを使用してページをダウンロードし、ファイルとして保存します

'''
Created on 2012-3-13
Get Page using GET method
Default using HTTP Protocol , http port 80
@author: xiaojay
'''
import socket
import statistics
import datetime
import threading
socket.setdefaulttimeout(statistics.timeout)
class Error404(Exception):
  '''Can not find the page.'''
  pass
class ErrorOther(Exception):
  '''Some other exception'''
  def __init__(self,code):
    #print 'Code :',code
    pass
class ErrorTryTooManyTimes(Exception):
  '''try too many times'''
  pass
def downPage(hostname ,filename , trytimes=0):
  try :
    #To avoid too many tries .Try times can not be more than max_try_times
    if trytimes >= statistics.max_try_times :
      raise ErrorTryTooManyTimes
  except ErrorTryTooManyTimes :
    return statistics.RESULTTRYTOOMANY,hostname+filename
  try:
    s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    #DNS cache
    if statistics.DNSCache.has_key(hostname):
      addr = statistics.DNSCache[hostname]
    else:
      addr = socket.gethostbyname(hostname)
      statistics.DNSCache[hostname] = addr
    #connect to http server ,default port 80
    s.connect((addr,80))
    msg = 'GET '+filename+' HTTP/1.0\r\n'
    msg += 'Host: '+hostname+'\r\n'
    msg += 'User-Agent:xiaojay\r\n\r\n'
    code = ''
    f = None
    s.sendall(msg)
    first = True
    while True:
      msg = s.recv(40960)
      if not len(msg):
        if f!=None:
          f.flush()
          f.close()
        break
      # Head information must be in the first recv buffer
      if first:
        first = False
        headpos = msg.index("\r\n\r\n")
        code,other = dealwithHead(msg[:headpos])
        if code=='200':
          #statistics.fetched_url += 1
          f = open('pages/'+str(abs(hash(hostname+filename))),'w')
          f.writelines(msg[headpos+4:])
        elif code=='301' or code=='302':
          #if code is 301 or 302 , try down again using redirect location
          if other.startswith("http") :
            hname, fname = parse(other)
            downPage(hname,fname,trytimes+1)#try again
          else :
            downPage(hostname,other,trytimes+1)
        elif code=='404':
          raise Error404
        else :
          raise ErrorOther(code)
      else:
        if f!=None :f.writelines(msg)
    s.shutdown(socket.SHUT_RDWR)
    s.close()
    return statistics.RESULTFETCHED,hostname+filename
  except Error404 :
    return statistics.RESULTCANNOTFIND,hostname+filename
  except ErrorOther:
    return statistics.RESULTOTHER,hostname+filename
  except socket.timeout:
    return statistics.RESULTTIMEOUT,hostname+filename
  except Exception, e:
    return statistics.RESULTOTHER,hostname+filename
def dealwithHead(head):
  '''deal with HTTP HEAD'''
  lines = head.splitlines()
  fstline = lines[0]
  code =fstline.split()[1]
  if code == '404' : return (code,None)
  if code == '200' : return (code,None)
  if code == '301' or code == '302' :
    for line in lines[1:]:
      p = line.index(':')
      key = line[:p]
      if key=='Location' :
        return (code,line[p+2:])
  return (code,None)
def parse(url):
  '''Parse a url to hostname+filename'''
  try:
    u = url.strip().strip('\n').strip('\r').strip('\t')
    if u.startswith('http://') :
      u = u[7:]
    elif u.startswith('https://'):
      u = u[8:]
    if u.find(':80')>0 :
      p = u.index(':80')
      p2 = p + 3
    else:
      if u.find('/')>0:
        p = u.index('/')
        p2 = p
      else:
        p = len(u)
        p2 = -1
    hostname = u[:p]
    if p2>0 :
      filename = u[p2:]
    else : filename = '/'
    return hostname, filename
  except Exception ,e:
    print "Parse wrong : " , url
    print e
def PrintDNSCache():
  '''print DNS dict'''
  n = 1
  for hostname in statistics.DNSCache.keys():
    print n,'\t',hostname, '\t',statistics.DNSCache[hostname]
    n+=1
def dealwithResult(res,url):
  '''Deal with the result of downPage'''
  statistics.total_url+=1
  if res==statistics.RESULTFETCHED :
    statistics.fetched_url+=1
    print statistics.total_url , '\t fetched :', url
  if res==statistics.RESULTCANNOTFIND :
    statistics.failed_url+=1
    print "Error 404 at : ", url
  if res==statistics.RESULTOTHER :
    statistics.other_url +=1
    print "Error Undefined at : ", url
  if res==statistics.RESULTTIMEOUT :
    statistics.timeout_url +=1
    print "Timeout ",url
  if res==statistics.RESULTTRYTOOMANY:
    statistics.trytoomany_url+=1
    print e ,"Try too many times at", url
if __name__=='__main__':
  print 'Get Page using GET method'

次に、スレッドを使用します前回の記事のプール 補助的にマルチスレッド下での並列クロールを実装し、上で書いたページをダウンロードする方法でurllib2とのパフォーマンスを比較してみます。

'''
Created on 2012-3-16
@author: xiaojay
'''
import fetchPage
import threadpool
import datetime
import statistics
import urllib2
'''one thread'''
def usingOneThread(limit):
  urlset = open("input.txt","r")
  start = datetime.datetime.now()
  for u in urlset:
    if limit <= 0 : break
    limit-=1
    hostname , filename = parse(u)
    res= fetchPage.downPage(hostname,filename,0)
    fetchPage.dealwithResult(res)
  end = datetime.datetime.now()
  print "Start at :\t" , start
  print "End at :\t" , end
  print "Total Cost :\t" , end - start
  print &#39;Total fetched :&#39;, statistics.fetched_url
&#39;&#39;&#39;threadpoll and GET method&#39;&#39;&#39;
def callbackfunc(request,result):
  fetchPage.dealwithResult(result[0],result[1])
def usingThreadpool(limit,num_thread):
  urlset = open("input.txt","r")
  start = datetime.datetime.now()
  main = threadpool.ThreadPool(num_thread)
  for url in urlset :
    try :
      hostname , filename = fetchPage.parse(url)
      req = threadpool.WorkRequest(fetchPage.downPage,args=[hostname,filename],kwds={},callback=callbackfunc)
      main.putRequest(req)
    except Exception:
      print Exception.message
  while True:
    try:
      main.poll()
      if statistics.total_url >= limit : break
    except threadpool.NoResultsPending:
      print "no pending results"
      break
    except Exception ,e:
      print e
  end = datetime.datetime.now()
  print "Start at :\t" , start
  print "End at :\t" , end
  print "Total Cost :\t" , end - start
  print &#39;Total url :&#39;,statistics.total_url
  print &#39;Total fetched :&#39;, statistics.fetched_url
  print &#39;Lost url :&#39;, statistics.total_url - statistics.fetched_url
  print &#39;Error 404 :&#39; ,statistics.failed_url
  print &#39;Error timeout :&#39;,statistics.timeout_url
  print &#39;Error Try too many times &#39; ,statistics.trytoomany_url
  print &#39;Error Other faults &#39;,statistics.other_url
  main.stop()
&#39;&#39;&#39;threadpool and urllib2 &#39;&#39;&#39;
def downPageUsingUrlib2(url):
  try:
    req = urllib2.Request(url)
    fd = urllib2.urlopen(req)
    f = open("pages3/"+str(abs(hash(url))),&#39;w&#39;)
    f.write(fd.read())
    f.flush()
    f.close()
    return url ,&#39;success&#39;
  except Exception:
    return url , None
def writeFile(request,result):
  statistics.total_url += 1
  if result[1]!=None :
    statistics.fetched_url += 1
    print statistics.total_url,&#39;\tfetched :&#39;, result[0],
  else:
    statistics.failed_url += 1
    print statistics.total_url,&#39;\tLost :&#39;,result[0],
def usingThreadpoolUrllib2(limit,num_thread):
  urlset = open("input.txt","r")
  start = datetime.datetime.now()
  main = threadpool.ThreadPool(num_thread)
  for url in urlset :
    try :
      req = threadpool.WorkRequest(downPageUsingUrlib2,args=[url],kwds={},callback=writeFile)
      main.putRequest(req)
    except Exception ,e:
      print e
  while True:
    try:
      main.poll()
      if statistics.total_url >= limit : break
    except threadpool.NoResultsPending:
      print "no pending results"
      break
    except Exception ,e:
      print e
  end = datetime.datetime.now()
  print "Start at :\t" , start
  print "End at :\t" , end
  print "Total Cost :\t" , end - start
  print &#39;Total url :&#39;,statistics.total_url
  print &#39;Total fetched :&#39;, statistics.fetched_url
  print &#39;Lost url :&#39;, statistics.total_url - statistics.fetched_url
  main.stop()
if __name__ ==&#39;__main__&#39;:
  &#39;&#39;&#39;too slow&#39;&#39;&#39;
  #usingOneThread(100)
  &#39;&#39;&#39;use Get method&#39;&#39;&#39;
  #usingThreadpool(3000,50)
  &#39;&#39;&#39;use urllib2&#39;&#39;&#39;
  usingThreadpoolUrllib2(3000,50)

実験分析:

実験データ:

larbin によってキャプチャされた 3000 の URL はメルカトル キュー モデルによって処理されました (C++ で実装しました。将来の機会) ランダムで代表的な URL のコレクション。 50 スレッドのスレッド プールを使用します。

実験環境: ubuntu10.04、良好なネットワーク、python2.6
ストレージ: 小さなファイル、各ページ、ストレージ用に 1 つのファイル PS: 学校のインターネット アクセスはトラフィックに基づいて課金されるため、Web クローラーを実行するとトラフィックが高くなります。 ！ ！ ！数日以内に、大規模な URL ダウンロード実験を実施し、数十万の URL で試す可能性があります。

実験結果:

Using urllib2

、usingThreadpoolUrllib2(3000,50)

Start at : 2012-03-16 22:18:20.956054End at : 2012-03 -16 22:22:15.203018

合計コスト: 0:03:54.246964

合計 URL: 3001
取得された合計: 2442
失われた URL: 559

ダウンロード ページの物理ストレージ サイズ: 84088kb

Threadpool(3000,50) を使用して独自の getPageUsingGet を使用してください

開始時間: 2012-03-16 22:23:40.206730

終了時刻: 2012-03-16 22:26:26.843563

合計コスト: 0:02:46.636833

合計 URL : 3002
合計取得数 : 2484
Lo st URL : 518
エラー 404 : 94
エラー タイムアウト : 312
エラー 試行回数が多すぎます 0
エラー その他の障害 112

ダウンロード ページの物理ストレージ サイズ: 87168kb

概要: 私が自分で書いたダウンロード ページ プログラムは非常に効率的で、失われるページが少なくなります。しかし、実際には、最適化できる箇所はまだたくさんあります。たとえば、ファイルが分散しすぎると、プログラムのパフォーマンスに大きなオーバーヘッドが発生することは間違いありません。はハッシュ命名を使用しますが、計算に関しては多くの問題が発生しますが、適切な戦略があれば、これらのオーバーヘッドは実際には省略できます。 DNS に加えて、Python に付属の DNS 解決を使用する必要はありません。デフォルトの DNS 解決は同期操作であり、DNS 解決は一般に時間がかかるため、マルチスレッドの非同期で実行できます。適切な DNS キャッシュと組み合わせることで、効率を大幅に向上させることができます。それだけでなく、実際のページ クローリング プロセス中には大量の URL が存在するため、それらを一度にメモリに保存することは不可能であり、特定の戦略またはアルゴリズムに従って合理的に割り当てられる必要があります。 つまり、コレクションページにはやるべきこと、最適化できることがまだたくさんあります。

以上がPython はマルチスレッドを使用して Web ページ情報をクロールしますの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。