Python多线程、异步＋多进程爬虫实现代码

概述安装Tornado省事点可以直接用grequests库，下面用的是tornado的异步client。异步用到了tornado，根据官方文档的例子修改得到一个简单的异步爬虫类。可以参考下最新的文档学习下。

安装Tornado
省事点可以直接用grequests库，下面用的是tornado的异步clIEnt。 异步用到了tornado，根据官方文档的例子修改得到一个简单的异步爬虫类。可以参考下最新的文档学习下。
pip install tornado

异步爬虫

#!/usr/bin/env python# -*- Coding:utf-8 -*-import timefrom datetime import timedeltafrom tornado import httpclIEnt,gen,ioloop,queuesimport tracebackclass AsySpIDer(object):  """A simple class of asynchronous spIDer."""  def __init__(self,urls,concurrency=10,**kwargs):    urls.reverse()    self.urls = urls    self.concurrency = concurrency    self._q = queues.Queue()    self._fetching = set()    self._fetched = set()  def fetch(self,url,**kwargs):    fetch = getattr(httpclIEnt.AsynchttpClIEnt(),'fetch')    return fetch(url,**kwargs)  def handle_HTML(self,HTML):    """handle HTML page"""    print(url)  def handle_response(self,response):    """inherit and rewrite this method"""    if response.code == 200:      self.handle_HTML(url,response.body)    elif response.code == 599:  # retry      self._fetching.remove(url)      self._q.put(url)  @gen.coroutine  def get_page(self,url):    try:      response = yIEld self.fetch(url)      print('######fetched %s' % url)    except Exception as e:      print('Exception: %s %s' % (e,url))      raise gen.Return(e)    raise gen.Return(response)  @gen.coroutine  def _run(self):    @gen.coroutine    def fetch_url():      current_url = yIEld self._q.get()      try:        if current_url in self._fetching:          return        print('fetching****** %s' % current_url)        self._fetching.add(current_url)        response = yIEld self.get_page(current_url)        self.handle_response(current_url,response)  # handle reponse        self._fetched.add(current_url)        for i in range(self.concurrency):          if self.urls:            yIEld self._q.put(self.urls.pop())      finally:        self._q.task_done()    @gen.coroutine    def worker():      while True:        yIEld fetch_url()    self._q.put(self.urls.pop())  # add first url    # Start workers,then wait for the work queue to be empty.    for _ in range(self.concurrency):      worker()    yIEld self._q.join(timeout=timedelta(seconds=300000))    assert self._fetching == self._fetched  def run(self):    io_loop = ioloop.Ioloop.current()    io_loop.run_sync(self._run)class MySpIDer(AsySpIDer):  def fetch(self,**kwargs):    """重写父类fetch方法可以添加cookies，headers，timeout等信息"""    cookies_str = "PHPSESSID=j1tt66a829IDnms56ppb70jri4; pspt=%7B%22ID%22%3A%2233153%22%2C%22pswd%22%3A%228835d2c1351d221b4ab016fbf9e8253f%22%2C%22_code%22%3A%22f779dcd011f4e2581c716d1e1b945861%22%7D; key=%E9%87%8D%E5%BA%86%E5%95%84%E6%9C%A8%E9%B8%9F%E7%BD%91%E7%BB%9C%E7%A7%91%E6%8A%80%E6%9C%89%E9%99%90%E5%85%AC%E5%8F%B8; think_language=zh-cn; SERVERID=a66d7d08fa1c8b2e37dbdc6ffff82d9e|1444973193|1444967835; CNZZDATA1254842228=1433864393-1442810831-%7C1444972138"  # 从浏览器拷贝cookie字符串    headers = {      'User-Agent': 'mozilla/5.0 (compatible; baIDuspIDer/2.0; +http://www.baIDu.com/search/spIDer.HTML)','cookie': cookies_str    }    return super(MySpIDer,self).fetch(  # 参数参考tornado文档      url,headers=headers,request_timeout=1    )  def handle_HTML(self,HTML):    print(url,HTML)def main():  urls = []  for page in range(1,100):    urls.append('http://www.baIDu.com?page=%s' % page)  s = MySpIDer(urls)  s.run()if __name__ == '__main__':  main()

可以继承这个类，塞一些url进去，然后重写handle_page处理得到的页面。

异步＋多进程爬虫
还可以再变态点，加个进程池，使用了multiprocessing模块。效率飕飕的,

#!/usr/bin/env python# -*- Coding:utf-8 -*-import timefrom multiprocessing import Poolfrom datetime import timedeltafrom tornado import httpclIEnt,queuesclass AsySpIDer(object):  """A simple class of asynchronous spIDer."""  def __init__(self,concurrency):    urls.reverse()    self.urls = urls    self.concurrency = concurrency    self._q = queues.Queue()    self._fetching = set()    self._fetched = set()  def handle_page(self,HTML):    filename = url.rsplit('/',1)[1]    with open(filename,'w+') as f:      f.write(HTML)  @gen.coroutine  def get_page(self,url):    try:      response = yIEld httpclIEnt.AsynchttpClIEnt().fetch(url)      print('######fetched %s' % url)    except Exception as e:      print('Exception: %s %s' % (e,url))      raise gen.Return('')    raise gen.Return(response.body)  @gen.coroutine  def _run(self):    @gen.coroutine    def fetch_url():      current_url = yIEld self._q.get()      try:        if current_url in self._fetching:          return        print('fetching****** %s' % current_url)        self._fetching.add(current_url)        HTML = yIEld self.get_page(current_url)        self._fetched.add(current_url)        self.handle_page(current_url,HTML)        for i in range(self.concurrency):          if self.urls:            yIEld self._q.put(self.urls.pop())      finally:        self._q.task_done()    @gen.coroutine    def worker():      while True:        yIEld fetch_url()    self._q.put(self.urls.pop())    # Start workers,then wait for the work queue to be empty.    for _ in range(self.concurrency):      worker()    yIEld self._q.join(timeout=timedelta(seconds=300000))    assert self._fetching == self._fetched  def run(self):    io_loop = ioloop.Ioloop.current()    io_loop.run_sync(self._run)def run_spIDer(beg,end):  urls = []  for page in range(beg,end):    urls.append('http://127.0.0.1/%s.htm' % page)  s = AsySpIDer(urls,10)  s.run()def main():  _st = time.time()  p = Pool()  all_num = 73000  num = 4  # number of cpu cores  per_num,left = divmod(all_num,num)  s = range(0,all_num,per_num)  res = []  for i in range(len(s)-1):    res.append((s[i],s[i+1]))  res.append((s[len(s)-1],all_num))  print res  for i in res:    p.apply_async(run_spIDer,args=(i[0],i[1],))  p.close()  p.join()  print time.time()-_stif __name__ == '__main__':  main()

多线程爬虫
线程池实现.

#!/usr/bin/env python# -*- Coding:utf-8 -*-import Queueimport sysimport requestsimport osimport threadingimport timeclass Worker(threading.Thread):  # 处理工作请求  def __init__(self,workQueue,resultQueue,**kwds):    threading.Thread.__init__(self,**kwds)    self.setDaemon(True)    self.workQueue = workQueue    self.resultQueue = resultQueue  def run(self):    while 1:      try:        callable,args,kwds = self.workQueue.get(False)  # get task        res = callable(*args,**kwds)        self.resultQueue.put(res)  # put result      except Queue.Empty:        breakclass WorkManager:  # 线程池管理,创建  def __init__(self,num_of_workers=10):    self.workQueue = Queue.Queue()  # 请求队列    self.resultQueue = Queue.Queue()  # 输出结果的队列    self.workers = []    self._recruitThreads(num_of_workers)  def _recruitThreads(self,num_of_workers):    for i in range(num_of_workers):      worker = Worker(self.workQueue,self.resultQueue)  # 创建工作线程      self.workers.append(worker)  # 加入到线程队列  def start(self):    for w in self.workers:      w.start()  def wait_for_complete(self):    while len(self.workers):      worker = self.workers.pop()  # 从池中取出一个线程处理请求      worker.join()      if worker.isAlive() and not self.workQueue.empty():        self.workers.append(worker)  # 重新加入线程池中    print 'All jobs were complete.'  def add_job(self,callable,*args,**kwds):    self.workQueue.put((callable,kwds))  # 向工作队列中加入请求  def get_result(self,**kwds):    return self.resultQueue.get(*args,**kwds)def download_file(url):  #print 'beg download',url  requests.get(url).textdef main():  try:    num_of_threads = int(sys.argv[1])  except:    num_of_threads = 10  _st = time.time()  wm = WorkManager(num_of_threads)  print num_of_threads  urls = ['http://www.baIDu.com'] * 1000  for i in urls:    wm.add_job(download_file,i)  wm.start()  wm.wait_for_complete()  print time.time() - _stif __name__ == '__main__':  main()

这三种随便一种都有很高的效率，但是这么跑会给网站服务器不小的压力，尤其是小站点，还是有点节 *** 为好。

您可能感兴趣的文章:基python实现多线程网页爬虫python实现爬虫统计学校BBS男女比例之多线程爬虫（二）Python多线程爬虫简单示例Python 爬虫学习笔记之多线程爬虫使用Python多线程爬虫爬取电影天堂资源Python 爬虫多线程详解及实例代码php与python实现的线程池多线程爬虫功能示例Python多线程爬虫实战_爬取糗事百科段子的实例Python3多线程爬虫实例讲解代码Python之多线程爬虫抓取网页图片的示例代码 总结

以上是内存溢出为你收集整理的Python多线程、异步＋多进程爬虫实现代码全部内容，希望文章能够帮你解决Python多线程、异步＋多进程爬虫实现代码所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错，欢迎将内存溢出网站推荐给程序员好友。