Python实现的异步代理爬虫及代理池

async def start(self):
 for rule in self._rules:
 parser = asyncio.ensure_future(self._parse_page(rule)) # 根据规则解析页面来获取代理
 logger.debug('{0} crawler started'.format(rule.__rule_name__))
 if not rule.use_phantomjs:
  await page_download(ProxyCrawler._url_generator(rule), self._pages, self._stop_flag) # 爬取代理网站的页面
 else:
  await page_download_phantomjs(ProxyCrawler._url_generator(rule), self._pages,
rule.phantomjs_load_flag, self._stop_flag) # 使用PhantomJS爬取
 await self._pages.join()
 parser.cancel()
 logger.debug('{0} crawler finished'.format(rule.__rule_name__))

import asyncio
from random import random
async def produce(queue, n):
 for x in range(1, n + 1):
 print('produce ', x)
 await asyncio.sleep(random())
 await queue.put(x) # 向queue中放入item
async def consume(queue):
 while 1:
 item = await queue.get() # 等待从queue中获取item
 print('consume ', item)
 await asyncio.sleep(random())
 queue.task_done() # 通知queue当前item处理完毕 
async def run(n):
 queue = asyncio.Queue()
 consumer = asyncio.ensure_future(consume(queue))
 await produce(queue, n) # 等待生产者结束
 await queue.join() # 阻塞直到queue不为空
 consumer.cancel() # 取消消费者任务，否则它会一直阻塞在get方法处
def aio_queue_run(n):
 loop = asyncio.get_event_loop()
 try:
 loop.run_until_complete(run(n)) # 持续运行event loop直到任务run(n)结束
 finally:
 loop.close()
if __name__ == '__main__':
 aio_queue_run(5)

produce 1
produce 2
consume 1
produce 3
produce 4
consume 2
produce 5
consume 3
consume 4
consume 5

async def page_download(urls, pages, flag):
 url_generator = urls
 async with aiohttp.ClientSession() as session:
 for url in url_generator:
  if flag.is_set():
  break
  await asyncio.sleep(uniform(delay - 0.5, delay + 1))
  logger.debug('crawling proxy web page {0}'.format(url))
  try:
  async with session.get(url, headers=headers, timeout=10) as response:
   page = await response.text()
   parsed = html.fromstring(decode_html(page)) # 使用bs4来辅助lxml解码网页:http://lxml.de/elementsoup.html#Using only the encoding detection
   await pages.put(parsed)
   url_generator.send(parsed) # 根据当前页面来获取下一页的地址
  except StopIteration:
  break
  except asyncio.TimeoutError:
  logger.error('crawling {0} timeout'.format(url))
  continue # TODO: use a proxy
  except Exception as e:
  logger.error(e)

async def _parse_proxy(self, rule, page):
 ips = page.xpath(rule.ip_xpath) # 根据xpath解析得到list类型的ip地址集合
 ports = page.xpath(rule.port_xpath) # 根据xpath解析得到list类型的ip地址集合
 if not ips or not ports:
 logger.warning('{2} crawler could not get ip(len={0}) or port(len={1}), please check the xpaths or network'.
  format(len(ips), len(ports), rule.__rule_name__))
 return
 proxies = map(lambda x, y: '{0}:{1}'.format(x.text.strip(), y.text.strip()), ips, ports)
 if rule.filters: # 根据过滤字段来过滤代理，如“高匿”、“透明”等
 filters = []
 for i, ft in enumerate(rule.filters_xpath):
  field = page.xpath(ft)
  if not field:
  logger.warning('{1} crawler could not get {0} field, please check the filter xpath'.
   format(rule.filters[i], rule.__rule_name__))
  continue
  filters.append(map(lambda x: x.text.strip(), field))
 filters = zip(*filters)
 selector = map(lambda x: x == rule.filters, filters)
 proxies = compress(proxies, selector)
for proxy in proxies:
await self._proxies.put(proxy) # 解析后的代理放入asyncio.Queue中

class CrawlerRuleBase(object, metaclass=CrawlerRuleMeta):
 start_url = None
 page_count = 0
 urls_format = None
 next_page_xpath = None
 next_page_host = ''
 use_phantomjs = False
 phantomjs_load_flag = None
 filters = ()
 ip_xpath = None
 port_xpath = None
 filters_xpath = ()

async def validate(self, proxies):
 logger.debug('validator started')
 while 1:
 proxy = await proxies.get()
 async with aiohttp.ClientSession() as session:
  try:
  real_proxy = 'http://' + proxy
  async with session.get(self.validate_url, proxy=real_proxy, timeout=validate_timeout) as resp:
   self._conn.put(proxy)
  except Exception as e:
  logger.error(e)
 proxies.task_done()

; supervisord.conf
[unix_http_server]
file=/tmp/supervisor.sock 

[inet_http_server]  
port=127.0.0.1:9001 

[supervisord]
logfile=/tmp/supervisord.log 
logfile_maxbytes=5MB 
logfile_backups=10  
loglevel=debug  
pidfile=/tmp/supervisord.pid 
nodaemon=false  
minfds=1024   
minprocs=200   

[rpcinterface:supervisor]
supervisor.rpcinterface_factory = supervisor.rpcinterface:make_main_rpcinterface

[supervisorctl]
serverurl=unix:///tmp/supervisor.sock

[program:proxyPool]
command=python /path/to/ProxyPool/run_proxypool.py  
redirect_stderr=true
stdout_logfile=NONE

[program:proxyValidator]
command=python /path/to/ProxyPool/run_proxyvalidator.py
redirect_stderr=true  
stdout_logfile=NONE

[program:proxyServer]
command=python /path/to/ProxyPool/run_proxyserver.py
autostart=false
redirect_stderr=true  
stdout_logfile=NONE