Python多线程、异步＋多进程爬虫实现代码
安装Tornado
省事点可以直接用grequests库，下面用的是tornado的异步client。异步用到了tornado，根据官方文档的例子修改得到一个简单的异步爬虫类。可以参考下最新的文档学习下。
pipinstalltornado
异步爬虫
#!/usr/bin/envpython
#-*-coding:utf-8-*-
importtime
fromdatetimeimporttimedelta
fromtornadoimporthttpclient,gen,ioloop,queues
importtraceback
classAsySpider(object):
"""Asimpleclassofasynchronousspider."""
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()
deffetch(self,url,**kwargs):
fetch=getattr(httpclient.AsyncHTTPClient(),'fetch')
returnfetch(url,**kwargs)
defhandle_html(self,url,html):
"""handlehtmlpage"""
print(url)
defhandle_response(self,url,response):
"""inheritandrewritethismethod"""
ifresponse.code==200:
self.handle_html(url,response.body)
elifresponse.code==599:#retry
self._fetching.remove(url)
self._q.put(url)
@gen.coroutine
defget_page(self,url):
try:
response=yieldself.fetch(url)
print('######fetched%s'%url)
exceptExceptionase:
print('Exception:%s%s'%(e,url))
raisegen.Return(e)
raisegen.Return(response)
@gen.coroutine
def_run(self):
@gen.coroutine
deffetch_url():
current_url=yieldself._q.get()
try:
ifcurrent_urlinself._fetching:
return
print('fetching******%s'%current_url)
self._fetching.add(current_url)
response=yieldself.get_page(current_url)
self.handle_response(current_url,response)#handlereponse
self._fetched.add(current_url)
foriinrange(self.concurrency):
ifself.urls:
yieldself._q.put(self.urls.pop())
finally:
self._q.task_done()
@gen.coroutine
defworker():
whileTrue:
yieldfetch_url()
self._q.put(self.urls.pop())#addfirsturl
#Startworkers,thenwaitfortheworkqueuetobeempty.
for_inrange(self.concurrency):
worker()
yieldself._q.join(timeout=timedelta(seconds=300000))
assertself._fetching==self._fetched
defrun(self):
io_loop=ioloop.IOLoop.current()
io_loop.run_sync(self._run)
classMySpider(AsySpider):
deffetch(self,url,**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
}
returnsuper(MySpider,self).fetch(#参数参考tornado文档
url,headers=headers,request_timeout=1
)
defhandle_html(self,url,html):
print(url,html)
defmain():
urls=[]
forpageinrange(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/envpython
#-*-coding:utf-8-*-
importtime
frommultiprocessingimportPool
fromdatetimeimporttimedelta
fromtornadoimporthttpclient,gen,ioloop,queues
classAsySpider(object):
"""Asimpleclassofasynchronousspider."""
def__init__(self,urls,concurrency):
urls.reverse()
self.urls=urls
self.concurrency=concurrency
self._q=queues.Queue()
self._fetching=set()
self._fetched=set()
defhandle_page(self,url,html):
filename=url.rsplit('/',1)[1]
withopen(filename,'w+')asf:
f.write(html)
@gen.coroutine
defget_page(self,url):
try:
response=yieldhttpclient.AsyncHTTPClient().fetch(url)
print('######fetched%s'%url)
exceptExceptionase:
print('Exception:%s%s'%(e,url))
raisegen.Return('')
raisegen.Return(response.body)
@gen.coroutine
def_run(self):
@gen.coroutine
deffetch_url():
current_url=yieldself._q.get()
try:
ifcurrent_urlinself._fetching:
return
print('fetching******%s'%current_url)
self._fetching.add(current_url)
html=yieldself.get_page(current_url)
self._fetched.add(current_url)
self.handle_page(current_url,html)
foriinrange(self.concurrency):
ifself.urls:
yieldself._q.put(self.urls.pop())
finally:
self._q.task_done()
@gen.coroutine
defworker():
whileTrue:
yieldfetch_url()
self._q.put(self.urls.pop())
#Startworkers,thenwaitfortheworkqueuetobeempty.
for_inrange(self.concurrency):
worker()
yieldself._q.join(timeout=timedelta(seconds=300000))
assertself._fetching==self._fetched
defrun(self):
io_loop=ioloop.IOLoop.current()
io_loop.run_sync(self._run)
defrun_spider(beg,end):
urls=[]
forpageinrange(beg,end):
urls.append('http://127.0.0.1/%s.htm'%page)
s=AsySpider(urls,10)
s.run()
defmain():
_st=time.time()
p=Pool()
all_num=73000
num=4#numberofcpucores
per_num,left=divmod(all_num,num)
s=range(0,all_num,per_num)
res=[]
foriinrange(len(s)-1):
res.append((s[i],s[i+1]))
res.append((s[len(s)-1],all_num))
printres
foriinres:
p.apply_async(run_spider,args=(i[0],i[1],))
p.close()
p.join()
printtime.time()-_st
if__name__=='__main__':
main()
多线程爬虫
线程池实现.
#!/usr/bin/envpython #-*-coding:utf-8-*- importQueue importsys importrequests importos importthreading importtime classWorker(threading.Thread):#处理工作请求 def__init__(self,workQueue,resultQueue,**kwds): threading.Thread.__init__(self,**kwds) self.setDaemon(True) self.workQueue=workQueue self.resultQueue=resultQueue defrun(self): while1: try: callable,args,kwds=self.workQueue.get(False)#gettask res=callable(*args,**kwds) self.resultQueue.put(res)#putresult exceptQueue.Empty: break classWorkManager:#线程池管理,创建 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): foriinrange(num_of_workers): worker=Worker(self.workQueue,self.resultQueue)#创建工作线程 self.workers.append(worker)#加入到线程队列 defstart(self): forwinself.workers: w.start() defwait_for_complete(self): whilelen(self.workers): worker=self.workers.pop()#从池中取出一个线程处理请求 worker.join() ifworker.isAlive()andnotself.workQueue.empty(): self.workers.append(worker)#重新加入线程池中 print'Alljobswerecomplete.' defadd_job(self,callable,*args,**kwds): self.workQueue.put((callable,args,kwds))#向工作队列中加入请求 defget_result(self,*args,**kwds): returnself.resultQueue.get(*args,**kwds) defdownload_file(url): #print'begdownload',url requests.get(url).text defmain(): try: num_of_threads=int(sys.argv[1]) except: num_of_threads=10 _st=time.time() wm=WorkManager(num_of_threads) printnum_of_threads urls=['http://www.baidu.com']*1000 foriinurls: wm.add_job(download_file,i) wm.start() wm.wait_for_complete() printtime.time()-_st if__name__=='__main__': main()
这三种随便一种都有很高的效率，但是这么跑会给网站服务器不小的压力，尤其是小站点，还是有点节操为好。