async def start(self): for rule in self._rules: parser = asyncio.ensure_future(self._parse_page(rule)) # 根據(jù)規(guī)則解析頁(yè)面來(lái)獲取代理 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) # 爬取代理網(wǎng)站的頁(yè)面 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 asynciofrom random import randomasync def produce(queue, n): for x in range(1, n + 1): print('produce ', x) await asyncio.sleep(random()) await queue.put(x) # 向queue中放入itemasync def consume(queue): while 1: item = await queue.get() # 等待從queue中獲取item print('consume ', item) await asyncio.sleep(random()) queue.task_done() # 通知queue當(dāng)前item處理完畢 async def run(n): queue = asyncio.Queue() consumer = asyncio.ensure_future(consume(queue)) await produce(queue, n) # 等待生產(chǎn)者結(jié)束 await queue.join() # 阻塞直到queue不為空 consumer.cancel() # 取消消費(fèi)者任務(wù)，否則它會(huì)一直阻塞在get方法處def aio_queue_run(n): loop = asyncio.get_event_loop() try: loop.run_until_complete(run(n)) # 持續(xù)運(yùn)行event loop直到任務(wù)run(n)結(jié)束 finally: loop.close()if __name__ == '__main__': aio_queue_run(5)

produce 1produce 2consume 1produce 3produce 4consume 2produce 5consume 3consume 4consume 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來(lái)輔助lxml解碼網(wǎng)頁(yè):http://lxml.de/elementsoup.html#Using only the encoding detection   await pages.put(parsed)   url_generator.send(parsed) # 根據(jù)當(dāng)前頁(yè)面來(lái)獲取下一頁(yè)的地址  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) # 根據(jù)xpath解析得到list類型的ip地址集合 ports = page.xpath(rule.port_xpath) # 根據(jù)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: # 根據(jù)過(guò)濾字段來(lái)過(guò)濾代理，如“高匿”、“透明”等 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=truestdout_logfile=NONE[program:proxyValidator]command=python /path/to/ProxyPool/run_proxyvalidator.pyredirect_stderr=true  stdout_logfile=NONE[program:proxyServer]command=python /path/to/ProxyPool/run_proxyserver.pyautostart=falseredirect_stderr=true  stdout_logfile=NONE