最近做了个爬取代理的爬虫,使用了python的aysncio及concurrent.futures的ThreadPoolExecutor(线程池)技术,最终完成了多线程下的异步抓取,在此mark下,以作备忘,代码在gitee上,是看到一位同道中人的go语言项目后比较感兴趣,于是用python加以改进并实现了相同的功能
基本思路就是配置好要爬取的免费代理地址,然后按照分页规则生成对应的地址,在组合成任务单元,提交给线程池,线程池则把任务分配给单一空闲线程,线程下把任务分为爬去数据,结果解析,有效性检验,存入数据库几个耗时操作,利用异步类将各操作组合起来,完成功能,篇幅限制就只列出主要代码了,可以当伪代码看下,希望对你有帮助
异步编程主要就是要把任务细分下来,分的好和分的坏差别是比较大的
废话不多说,上代码:
1.异步任务类
import asyncio,requestsfrom db.mysql import db #自己封装的sql包import pymysql.errclass asyncWorker(object): loop = None #事件循环 threadId = None #线程id def __init__(self,loop,tid): self.loop = loop self.threadId = tid ''' 协程任务,爬取到的结果入库 ''' async def execRes(self,sign,res): print('Thread#%d task#%d %s start' % (self.threadId,sign,json.dumps(res))) check = await self.check(res,sign) if check: await self.insertDB(**res,sign=sign) #爬取到的ip通过验证后插入数据库 ''' 协程任务,对爬取到的结果进行检查 ''' async def check(self,proxyData,sign): sign = self.ipVerify(**proxyData,sign=sign) return sign ''' 代理ip验证方法 ''' def ipVerify(self, ip , port , protocol,sign): url = protocol+'://httpbin.org/get' headers = {'User-Agent':'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.87 Safari/537.36'} if protocol in ['socks4','socks5']: url = 'https://httpbin.org/get' proxy = {'https':protocol+'://'+ip+':'+port} elif protocol in ['anonymous','distorting','transparent']: url = 'https://httpbin.org/get' proxy = {'https':'https://'+ip+':'+port} else: proxy = { protocol:protocol+'://'+ip+':'+port} try: res = requests.get(url,headers=headers,proxies=proxy,timeout=10) res.encoding = 'utf-8' res.raise_for_status() print('Thread#%d task#%d verify %s://%s:%s httpcode:%s' % (self.threadId,sign,protocol,ip,port,res.status_code)) return True except requests.exceptions.RequestException as e: print('Thread#%d task#%d verify %s' % (self.threadId,sign,e)) # AppLog('verify').error(e) return False ''' 协程任务,入库操作 ''' async def insertDB(self,ip,port,protocol,sign): try: database = db() database.insert(ip,port,protocol) print('Thread#%d task#%d insertDB ip:%s port:%s protocal:%s' % (self.threadId,sign,ip,port,protocol)) except pymysql.MySQLError as e: print('Thread#%d task#%d insertDB %s' % (self.threadId,sign,e)) ''' 并行执行方法 ''' def parallelrun(self,*task): asyncio.set_event_loop(self.loop) #设置对应线程下的事件循环 # loop = asyncio.get_event_loop() #如果不使用多线程,可以直接用这个来获取事件循环,上面的loop就可以删掉了 self.loop.run_until_complete(asyncio.gather(*task)) #事件循环直到所有任务完成 self.loop.close() #关闭事件循环
2.主函数线程池
import asyncio from asyncTask import asyncWorker import concurrent.futures #这是python封装的异步模块(python3.x内置),简单易用,人生苦短,我用python import time ''' 调度任务封装,线程下的工作单元 param: url-爬取代理的网址;parse-结果解析handle;loop-事件循环 ''' def work_url(tid,url,parse,loop): start = time.time() s = getter(url) #爬取网页的原始内容 res = s.getWeb(parse) #使用对应的解析函数对原始数据进行解析 worker = asyncWorker(loop,tid) #调用异步任务类 task = (worker.execRes(res.index(item),item) for item in res) #构造任务队列 worker.parallelrun(*task) #异步执行 end = time.time() print('[thread_id:%d]%s:%s page has been completed usage time:[%s]' % (tid,parse,url,end-start)) return '%r has been completed' % url urlIter = map(lambda item: {'parse':item[0],'url':item[1]},cfg.items('url')) #这里是爬取地址的生成器,cfg是配置类 #多线程,正片开始 with concurrent.futures.ThreadPoolExecutor(15) as exector: #先开15个线程试试,哈哈 future_to_url = {} #这个字典是用来存储对应的url和task的关联关系的,一一映射 channel = 1 #线程编号,就是asyncWorker中的threadId for item in urlIter: if item['parse'] == 'proxylists': for param in map(lambda page: {'url':item['url'].format(page=page),'parse':'proxylists','loop':asyncio.new_event_loop()}, range(10)): task = {exector.submit(work_url,tid=channel,**param):param['url']} #调用上面的工作单元并传参,提交给线程池 future_to_url.update(task) #更新关系字典 channel += 1 else: item['loop'] = asyncio.new_event_loop() #这里使用asyncio来新建事件循环 task = {exector.submit(work_url,tid=channel,**item):item['url']} future_to_url.update(task) channel += 1 for future in concurrent.futures.as_completed(future_to_url): url = future_to_url[future] try: data = future.result() #获取上面的工作单元的返回值 except Exception as e: print('%r generated an exception: %s' % (url,e)) else: print('[%r] %s' % (url,data)) exit()
备注:
1.python的多线程无法利用多核,只是靠cpu的时间分配造成的并发的错觉,当然,可以换用jython解释器来利用多核,默认的cython是存在GIL全局锁的
2.协程是一种用户态的线程,由用户来发出中断请求以及处理响应,省去了线程的切换开销,并且所有的处理是在同一个线程下,所以也不会造成读写冲突,省去了加锁操作
3.asyncio的async关键字等同于@asyncio.coroutine,await相当于yield from,两者区别就是后者的兼容性更好一些
4.吐槽一下,还是swoole的异步更方便
作者:_luka_
链接:https://www.jianshu.com/p/c5166df8e3b7