python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python多线程抓取网页

Python实现多线程抓取网页功能实例详解

作者:糖拌咸鱼

这篇文章主要介绍了Python实现多线程抓取网页功能,结合具体实例形式详细分析了Python多线程编程的相关操作技巧与注意事项,并附带demo实例给出了多线程抓取网页的实现方法,需要的朋友可以参考下

本文实例讲述了Python实现多线程抓取网页功能。分享给大家供大家参考,具体如下:

最近,一直在做网络爬虫相关的东西。 看了一下开源C++写的larbin爬虫,仔细阅读了里面的设计思想和一些关键技术的实现。

1、larbin的URL去重用的很高效的bloom filter算法;
2、DNS处理,使用的adns异步的开源组件;
3、对于url队列的处理,则是用部分缓存到内存,部分写入文件的策略。
4、larbin对文件的相关操作做了很多工作
5、在larbin里有连接池,通过创建套接字,向目标站点发送HTTP协议中GET方法,获取内容,再解析header之类的东西
6、大量描述字,通过poll方法进行I/O复用,很高效
7、larbin可配置性很强
8、作者所使用的大量数据结构都是自己从最底层写起的,基本没用STL之类的东西
......

还有很多,以后有时间在好好写篇文章,总结下。

这两天,用python写了个多线程下载页面的程序,对于I/O密集的应用而言,多线程显然是个很好的解决方案。刚刚写过的线程池,也正好可以利用上了。其实用python爬取页面非常简单,有个urllib2的模块,使用起来很方便,基本两三行代码就可以搞定。虽然使用第三方模块,可以很方便的解决问题,但是对个人的技术积累而言没有什么好处,因为关键的算法都是别人实现的,而不是你自己实现的,很多细节的东西,你根本就无法了解。 我们做技术的,不能一味的只是用别人写好的模块或是api,要自己动手实现,才能让自己学习得更多。

我决定从socket写起,也是去封装GET协议,解析header,而且还可以把DNS的解析过程单独处理,例如DNS缓存一下,所以这样自己写的话,可控性更强,更有利于扩展。对于timeout的处理,我用的全局的5秒钟的超时处理,对于重定位(301or302)的处理是,最多重定位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 'Total fetched :', statistics.fetched_url
'''threadpoll and GET method'''
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 'Total url :',statistics.total_url
  print 'Total fetched :', statistics.fetched_url
  print 'Lost url :', statistics.total_url - statistics.fetched_url
  print 'Error 404 :' ,statistics.failed_url
  print 'Error timeout :',statistics.timeout_url
  print 'Error Try too many times ' ,statistics.trytoomany_url
  print 'Error Other faults ',statistics.other_url
  main.stop()
'''threadpool and urllib2 '''
def downPageUsingUrlib2(url):
  try:
    req = urllib2.Request(url)
    fd = urllib2.urlopen(req)
    f = open("pages3/"+str(abs(hash(url))),'w')
    f.write(fd.read())
    f.flush()
    f.close()
    return url ,'success'
  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,'\tfetched :', result[0],
  else:
    statistics.failed_url += 1
    print statistics.total_url,'\tLost :',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 'Total url :',statistics.total_url
  print 'Total fetched :', statistics.fetched_url
  print 'Lost url :', statistics.total_url - statistics.fetched_url
  main.stop()
if __name__ =='__main__':
  '''too slow'''
  #usingOneThread(100)
  '''use Get method'''
  #usingThreadpool(3000,50)
  '''use urllib2'''
  usingThreadpoolUrllib2(3000,50)

实验分析:

实验数据:larbin抓取下来的3000条url,经过Mercator队列模型(我用c++实现的,以后有机会发个blog)处理后的url集合,具有随机和代表性。使用50个线程的线程池。
实验环境:ubuntu10.04,网络较好,python2.6
存储:小文件,每个页面,一个文件进行存储
PS:由于学校上网是按流量收费的,做网络爬虫,灰常费流量啊!!!过几天,可能会做个大规模url下载的实验,用个几十万的url试试。

实验结果:

使用urllib2 ,usingThreadpoolUrllib2(3000,50)

Start at :    2012-03-16 22:18:20.956054
End at :    2012-03-16 22:22:15.203018
Total Cost :    0:03:54.246964
Total url : 3001
Total fetched : 2442
Lost url : 559

下载页面的物理存储大小:84088kb

使用自己的getPageUsingGet ,usingThreadpool(3000,50)

Start at :    2012-03-16 22:23:40.206730
End at :    2012-03-16 22:26:26.843563
Total Cost :    0:02:46.636833
Total url : 3002
Total fetched : 2484
Lost url : 518
Error 404 : 94
Error timeout : 312
Error Try too many times  0
Error Other faults  112

下载页面的物理存储大小:87168kb

小结: 自己写的下载页面程序,效率还是很不错的,而且丢失的页面也较少。但其实自己考虑一下,还是有很多地方可以优化的,比如文件过于分散,过多的小文件创建和释放定会产生不小的性能开销,而且程序里用的是hash命名,也会产生很多的计算,如果有好的策略,其实这些开销都是可以省略的。另外DNS,也可以不使用python自带的DNS解析,因为默认的DNS解析都是同步的操作,而DNS解析一般比较耗时,可以采取多线程的异步的方式进行,再加以适当的DNS缓存很大程度上可以提高效率。不仅如此,在实际的页面抓取过程中,会有大量的url ,不可能一次性把它们存入内存,而应该按照一定的策略或是算法进行合理的分配。 总之,采集页面要做的东西以及可以优化的东西,还有很多很多。

附:demo源码点击此处本站下载

更多关于Python相关内容感兴趣的读者可查看本站专题:《Python进程与线程操作技巧总结》、《Python Socket编程技巧总结》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》、《Python入门与进阶经典教程》及《Python文件与目录操作技巧汇总

希望本文所述对大家Python程序设计有所帮助。

您可能感兴趣的文章:
阅读全文