前言书接上回在上篇中我们讲解了 Tornado 异步队列tornado.queues的核心运行原理以及异步生产者 - 消费者模式的基础实现思路。本篇我们将补全完整的可运行代码并逐点拆解整个异步爬虫的设计细节帮助大家彻底吃透这套经典异步架构的实现精髓。一、完整可运行代码先给出基于 Tornado 异步队列实现的全站爬虫完整代码在上篇核心逻辑的基础上补全了 worker 协程管理、任务超时控制、优雅退出机制等工业级实现细节#!/usr/bin/env python3 import asyncio import time from datetime import timedelta from html.parser import HTMLParser from urllib.parse import urljoin, urldefrag from tornado import gen, httpclient, queues base_url http://www.tornadoweb.org/en/stable/ concurrency 10 async def get_links_from_url(url): 抓取指定URL页面解析并返回页面内所有标准化后的链接 去除URL锚点转换为绝对路径避免重复抓取 response await httpclient.AsyncHTTPClient().fetch(url) print(ffetched {url}) html response.body.decode(errorsignore) return [urljoin(url, remove_fragment(new_url)) for new_url in get_links(html)] def remove_fragment(url): 去除URL中的锚点片段 pure_url, frag urldefrag(url) return pure_url def get_links(html): 解析HTML提取所有a标签的href链接 class URLSeeker(HTMLParser): def __init__(self): super().__init__() self.urls [] def handle_starttag(self, tag, attrs): href dict(attrs).get(href) if href and tag a: self.urls.append(href) url_seeker URLSeeker() url_seeker.feed(html) return url_seeker.urls async def main(): q queues.Queue() start time.time() # 分别记录抓取中、已抓取、抓取失败的URL用于去重与统计 fetching, fetched, dead set(), set(), set() async def fetch_url(current_url): 单个URL的抓取与链接解析逻辑生产者逻辑 if current_url in fetching: return print(ffetching {current_url}) fetching.add(current_url) urls await get_links_from_url(current_url) fetched.add(current_url) # 将符合域名规则的新链接放入队列作为新任务 for new_url in urls: if new_url.startswith(base_url): await q.put(new_url) async def worker(): 消费者协程持续从队列取任务执行 async for url in q: # 哨兵值收到None则退出协程 if url is None: return try: await fetch_url(url) except Exception as e: print(fException: {e} | {url}) dead.add(url) finally: # 标记当前任务处理完成