最近做了个爬取代理的爬虫，使用了python的aysncio及concurrent.futures的ThreadPoolExecutor（线程池）技术，最终完成了多线程下的异步抓取，在此mark下，以作备忘，代码在gitee上，是看到一位同道中人的go语言项目后比较感兴趣，于是用python加以改进并实现了相同的功能
基本思路就是配置好要爬取的免费代理地址，然后按照分页规则生成对应的地址，在组合成任务单元，提交给线程池，线程池则把任务分配给单一空闲线程，线程下把任务分为爬去数据，结果解析，有效性检验，存入数据库几个耗时操作，利用异步类将各操作组合起来，完成功能，篇幅限制就只列出主要代码了，可以当伪代码看下，希望对你有帮助
异步编程主要就是要把任务细分下来，分的好和分的坏差别是比较大的
废话不多说，上代码：
1.异步任务类
import asyncio,requests
from db.mysql import db #自己封装的sql包
import pymysql.err
class 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的异步更方便
