当前位置: 开发笔记 > 编程语言 > 正文

Scrapy源码剖析：Scrapy如何完成抓取任务？

作者：mobiledu2502861197 | 来源：互联网 | 2023-08-11 18:30

上一篇文章：Scrapy源码剖析：Scrapy有哪些核心组件？我们已经分析了Scrapy核心组件的主要职责，以及它们在初始化

上一篇文章&＃xff1a;Scrapy源码剖析&＃xff1a;Scrapy有哪些核心组件&＃xff1f;我们已经分析了 Scrapy 核心组件的主要职责&＃xff0c;以及它们在初始化时都完成了哪些工作。

这篇文章就让我们来看一下&＃xff0c;也是 Scrapy 最核心的抓取流程是如何运行的&＃xff0c;它是如何调度各个组件&＃xff0c;完成整个抓取工作的。

运行入口

还是回到最初的入口&＃xff0c;在Scrapy源码剖析&＃xff1a;Scrapy是如何运行起来的&＃xff1f;这篇文章中我们已经详细分析过了&＃xff0c;在执行 Scrapy 命令时&＃xff0c;主要经过以下几步&＃xff1a;

调用 cmdline.py 的 execute 方法
找到对应的 命令实例 解析命令行
构建 CrawlerProcess 实例&＃xff0c;调用 crawl 和 start 方法开始抓取

而 crawl 方法最终是调用了 Cralwer 实例的 crawl&＃xff0c;这个方法最终把控制权交给了Engine&＃xff0c;而 start 方法注册好协程池&＃xff0c;就开始异步调度执行了。

我们来看 Cralwer 的 crawl 方法&＃xff1a;

&＃64;defer.inlineCallbacks def crawl(self, *args, **kwargs):assert not self.crawling, "Crawling already taking place"self.crawling &＃61; Truetry:# 创建爬虫实例self.spider &＃61; self._create_spider(*args, **kwargs)# 创建引擎self.engine &＃61; self._create_engine()# 调用spider的start_requests 获取种子URLstart_requests &＃61; iter(self.spider.start_requests())# 调用engine的open_spider 交由引擎调度yield self.engine.open_spider(self.spider, start_requests)yield defer.maybeDeferred(self.engine.start)except Exception:if six.PY2:exc_info &＃61; sys.exc_info()self.crawling &＃61; Falseif self.engine is not None:yield self.engine.close()if six.PY2:six.reraise(*exc_info)raise

这里首先会创建出爬虫实例&＃xff0c;然后创建引擎&＃xff0c;之后调用了 spider 的 start_requests 方法&＃xff0c;这个方法就是我们平时写的最多爬虫类的父类&＃xff0c;它在 spiders/__init__.py 中定义&＃xff1a;

def start_requests(self):# 根据定义好的start_urls属性生成种子URL对象for url in self.start_urls:yield self.make_requests_from_url(url)def make_requests_from_url(self, url):# 构建Request对象return Request(url, dont_filter&＃61;True)构建请求

通过上面这段代码&＃xff0c;我们能看到&＃xff0c;平时我们必须要定义的 start_urls 属性&＃xff0c;原来就是在这里用来构建 Request 的&＃xff0c;来看 Request 的定义&＃xff1a;

class Request(object_ref):def __init__(self, url, callback&＃61;None, method&＃61;&＃39;GET&＃39;, headers&＃61;None, body&＃61;None,COOKIEs&＃61;None, meta&＃61;None, encoding&＃61;&＃39;utf-8&＃39;, priority&＃61;0,dont_filter&＃61;False, errback&＃61;None):# 编码self._encoding &＃61; encoding# 请求方法self.method &＃61; str(method).upper()# 设置urlself._set_url(url)# 设置bodyself._set_body(body)assert isinstance(priority, int), "Request priority not an integer: %r" % priority# 优先级self.priority &＃61; priorityassert callback or not errback, "Cannot use errback without a callback"# 回调函数self.callback &＃61; callback# 异常回调函数self.errback &＃61; errback# COOKIEsself.COOKIEs &＃61; COOKIEs or {}# 构建Headerself.headers &＃61; Headers(headers or {}, encoding&＃61;encoding)# 是否需要过滤self.dont_filter &＃61; dont_filter# 附加信息self._meta &＃61; dict(meta) if meta else None

Request 对象比较简单&＃xff0c;就是封装了请求参数、请求方法、回调以及可附加的属性信息。

当然&＃xff0c;你也可以在子类中重写 start_requests 和 make_requests_from_url 这 2 个方法&＃xff0c;用来自定义逻辑构建种子请求。

引擎调度

再回到 crawl 方法&＃xff0c;构建好种子请求对象后&＃xff0c;调用了 engine 的 open_spider&＃xff1a;

&＃64;defer.inlineCallbacks def open_spider(self, spider, start_requests&＃61;(), close_if_idle&＃61;True):assert self.has_capacity(), "No free spider slot when opening %r" % \spider.namelogger.info("Spider opened", extra&＃61;{&＃39;spider&＃39;: spider})# 注册_next_request调度方法循环调度nextcall &＃61; CallLaterOnce(self._next_request, spider)# 初始化schedulerscheduler &＃61; self.scheduler_cls.from_crawler(self.crawler)# 调用爬虫中间件处理种子请求start_requests &＃61; yield self.scraper.spidermw.process_start_requests(start_requests, spider)# 封装Slot对象slot &＃61; Slot(start_requests, close_if_idle, nextcall, scheduler)self.slot &＃61; slotself.spider &＃61; spider# 调用scheduler的openyield scheduler.open(spider)# 调用scrapyer的openyield self.scraper.open_spider(spider)# 调用stats的openself.crawler.stats.open_spider(spider)yield self.signals.send_catch_log_deferred(signals.spider_opened, spider&＃61;spider)# 发起调度slot.nextcall.schedule()slot.heartbeat.start(5)

在这里首先构建了一个 CallLaterOnce&＃xff0c;之后把 _next_request 方法注册了进去&＃xff0c;看此类的实现&＃xff1a;

class CallLaterOnce(object):# 在twisted的reactor中循环调度一个方法def __init__(self, func, *a, **kw):self._func &＃61; funcself._a &＃61; aself._kw &＃61; kwself._call &＃61; Nonedef schedule(self, delay&＃61;0):# 上次发起调度才可再次继续调度if self._call is None:# 注册self到callLater中self._call &＃61; reactor.callLater(delay, self)def cancel(self):if self._call:self._call.cancel()def __call__(self):# 上面注册的是self 所以会执行__call__self._call &＃61; Nonereturn self._func(*self._a, **self._kw)

这里封装了循环执行的方法类&＃xff0c;并且注册的方法会在 twisted 的 reactor 中异步执行&＃xff0c;以后执行只需调用 schedule&＃xff0c;就会注册 self 到 reactor 的 callLater 中&＃xff0c;然后它会执行 __call__ 方法&＃xff0c;最终执行的就是我们注册的方法。

而这里我们注册的方法就是引擎的 _next_request&＃xff0c;也就是说&＃xff0c;此方法会循环调度&＃xff0c;直到程序退出。

之后调用了爬虫中间件的 process_start_requests 方法&＃xff0c;你可以定义多个自己的爬虫中间件&＃xff0c;每个类都重写此方法&＃xff0c;爬虫在调度之前会分别调用你定义好的爬虫中间件&＃xff0c;来处理初始化请求&＃xff0c;你可以进行过滤、加工、筛选以及你想做的任何逻辑。

这样做的好处就是&＃xff0c;把想做的逻辑拆分成多个中间件&＃xff0c;每个中间件功能独立&＃xff0c;而且维护起来更加清晰。

调度器

接下来就要开始调度任务了&＃xff0c;这里首先调用了 Scheduler 的 open&＃xff1a;

def open(self, spider):self.spider &＃61; spider# 实例化优先级队列self.mqs &＃61; self.pqclass(self._newmq)# 如果定义了dqdir则实例化基于磁盘的队列self.dqs &＃61; self._dq() if self.dqdir else None# 调用请求指纹过滤器的open方法return self.df.open()def _dq(self):# 实例化磁盘队列activef &＃61; join(self.dqdir, &＃39;active.json&＃39;)if exists(activef):with open(activef) as f:prios &＃61; json.load(f)else:prios &＃61; ()q &＃61; self.pqclass(self._newdq, startprios&＃61;prios)if q:logger.info("Resuming crawl (%(queuesize)d requests scheduled)",{&＃39;queuesize&＃39;: len(q)}, extra&＃61;{&＃39;spider&＃39;: self.spider})return q

在 open 方法中&＃xff0c;调度器会实例化出优先级队列&＃xff0c;以及根据 dqdir是否配置&＃xff0c;决定是否使用磁盘队列&＃xff0c;最后调用了请求指纹过滤器的 open 方法&＃xff0c;这个方法在父类 BaseDupeFilter 中定义&＃xff1a;

class BaseDupeFilter(object):# 过滤器基类,子类可重写以下方法&＃64;classmethoddef from_settings(cls, settings):return cls()def request_seen(self, request):# 请求过滤return Falsedef open(self):# 可重写完成过滤器的初始化工作passdef close(self, reason):# 可重写完成关闭过滤器工作passdef log(self, request, spider):pas

请求过滤器提供了请求过滤的具体实现方式&＃xff0c;Scrapy 默认提供了 RFPDupeFilter 过滤器实现过滤重复请求的逻辑&＃xff0c;这里先对这个类有个了解&＃xff0c;后面会讲具体是如何过滤重复请求的。

Scraper

再之后就调用 Scraper 的 open_spider 方法&＃xff0c;在之前的文章中我们提到过&＃xff0c;Scraper 类是连接 Engine、Spider、Item Pipeline 这 3 个组件的桥梁&＃xff1a;

&＃64;defer.inlineCallbacks def open_spider(self, spider):self.slot &＃61; Slot()# 调用所有pipeline的open_spideryield self.itemproc.open_spider(spider)

这里的主要逻辑是 Scraper 调用所有 Pipeline 的 open_spider 方法&＃xff0c;如果我们定义了多个 Pipeline 输出类&＃xff0c;可以重写 open_spider 完成每个 Pipeline 在输出前的初始化工作。

循环调度

调用了一系列组件的 open 方法后&＃xff0c;最后调用了 nextcall.schedule() 开始调度&＃xff0c;也就是循环执行在上面注册的 _next_request 方法&＃xff1a;

def _next_request(self, spider):# 此方法会循环调度slot &＃61; self.slotif not slot:return# 暂停if self.paused:return# 是否等待while not self._needs_backout(spider):# 从scheduler中获取request# 注意&＃xff1a;第一次获取时&＃xff0c;是没有的&＃xff0c;也就是会break出来# 从而执行下面的逻辑if not self._next_request_from_scheduler(spider):break# 如果start_requests有数据且不需要等待if slot.start_requests and not self._needs_backout(spider):try:# 获取下一个种子请求request &＃61; next(slot.start_requests)except StopIteration:slot.start_requests &＃61; Noneexcept Exception:slot.start_requests &＃61; Nonelogger.error(&＃39;Error while obtaining start requests&＃39;,exc_info&＃61;True, extra&＃61;{&＃39;spider&＃39;: spider})else:# 调用crawl,实际是把request放入scheduler的队列中self.crawl(request, spider)# 空闲则关闭spiderif self.spider_is_idle(spider) and slot.close_if_idle:self._spider_idle(spider)def _needs_backout(self, spider):# 是否需要等待&＃xff0c;取决4个条件# 1. Engine是否stop# 2. slot是否close# 3. downloader下载超过预设# 4. scraper处理response超过预设slot &＃61; self.slotreturn not self.running \or slot.closing \or self.downloader.needs_backout() \or self.scraper.slot.needs_backout()def _next_request_from_scheduler(self, spider):slot &＃61; self.slot# 从scheduler拿出下个requestrequest &＃61; slot.scheduler.next_request()if not request:return# 下载d &＃61; self._download(request, spider)# 注册成功、失败、出口回调方法d.addBoth(self._handle_downloader_output, request, spider)d.addErrback(lambda f: logger.info(&＃39;Error while handling downloader output&＃39;,exc_info&＃61;failure_to_exc_info(f),extra&＃61;{&＃39;spider&＃39;: spider}))d.addBoth(lambda _: slot.remove_request(request))d.addErrback(lambda f: logger.info(&＃39;Error while removing request from slot&＃39;,exc_info&＃61;failure_to_exc_info(f),extra&＃61;{&＃39;spider&＃39;: spider}))d.addBoth(lambda _: slot.nextcall.schedule())d.addErrback(lambda f: logger.info(&＃39;Error while scheduling new request&＃39;,exc_info&＃61;failure_to_exc_info(f),extra&＃61;{&＃39;spider&＃39;: spider}))return ddef crawl(self, request, spider):assert spider in self.open_spiders, \"Spider %r not opened when crawling: %s" % (spider.name, request)# request放入scheduler队列&＃xff0c;调用nextcall的scheduleself.schedule(request, spider)self.slot.nextcall.schedule()def schedule(self, request, spider):self.signals.send_catch_log(signal&＃61;signals.request_scheduled,request&＃61;request, spider&＃61;spider)# 调用scheduler的enqueue_request&＃xff0c;把request放入scheduler队列if not self.slot.scheduler.enqueue_request(request):self.signals.send_catch_log(signal&＃61;signals.request_dropped,request&＃61;request, spider&＃61;spider)

_next_request 方法首先调用 _needs_backout 检查是否需要等待&＃xff0c;等待的条件有以下几种情况&＃xff1a;

引擎是否主动关闭
Slot是否关闭
下载器在网络下载时是否超过预设参数
Scraper处理输出是否超过预设参数

如果不需要等待&＃xff0c;则调用 _next_request_from_scheduler&＃xff0c;此方法从名字上就能看出&＃xff0c;主要是从 Schduler 中获取 Request。

这里要注意&＃xff0c;在第一次调用此方法时&＃xff0c;Scheduler 中是没有放入任何 Request 的&＃xff0c;这里会直接break 出来&＃xff0c;执行下面的逻辑&＃xff0c;而下面就会调用 crawl 方法&＃xff0c;实际是把请求放到 Scheduler 的请求队列&＃xff0c;放入队列的过程会经过请求过滤器校验是否重复。

下次再调用 _next_request_from_scheduler 时&＃xff0c;就能从 Scheduler 中获取到下载请求&＃xff0c;然后执行下载动作。

先来看第一次调度&＃xff0c;执行 crawl&＃xff1a;

def crawl(self, request, spider):assert spider in self.open_spiders, \"Spider %r not opened when crawling: %s" % (spider.name, request)# 放入Scheduler队列self.schedule(request, spider)# 进行下一次调度self.slot.nextcall.schedule()def schedule(self, request, spider):self.signals.send_catch_log(signal&＃61;signals.request_scheduled,request&＃61;request, spider&＃61;spider)# 放入Scheduler队列if not self.slot.scheduler.enqueue_request(request):self.signals.send_catch_log(signal&＃61;signals.request_dropped,request&＃61;request, spider&＃61;spider)

调用引擎的 crawl 实际就是把请求放入 Scheduler 的队列中&＃xff0c;下面看请求是如何入队列的。

请求入队

Scheduler 请求入队方法&＃xff1a;

def enqueue_request(self, request):# 请求入队若请求过滤器验证重复返回Falseif not request.dont_filter and self.df.request_seen(request):self.df.log(request, self.spider)return False# 磁盘队列是否入队成功dqok &＃61; self._dqpush(request)if dqok:self.stats.inc_value(&＃39;scheduler/enqueued/disk&＃39;, spider&＃61;self.spider)else:# 没有定义磁盘队列则使用内存队列self._mqpush(request)self.stats.inc_value(&＃39;scheduler/enqueued/memory&＃39;, spider&＃61;self.spider)self.stats.inc_value(&＃39;scheduler/enqueued&＃39;, spider&＃61;self.spider)return Truedef _dqpush(self, request):# 是否定义磁盘队列if self.dqs is None:returntry:# Request对象转dictreqd &＃61; request_to_dict(request, self.spider)# 放入磁盘队列self.dqs.push(reqd, -request.priority)except ValueError as e: # non serializable requestif self.logunser:msg &＃61; ("Unable to serialize request: %(request)s - reason:"" %(reason)s - no more unserializable requests will be"" logged (stats being collected)")logger.warning(msg, {&＃39;request&＃39;: request, &＃39;reason&＃39;: e},exc_info&＃61;True, extra&＃61;{&＃39;spider&＃39;: self.spider})self.logunser &＃61; Falseself.stats.inc_value(&＃39;scheduler/unserializable&＃39;,spider&＃61;self.spider)returnelse:return Truedef _mqpush(self, request):# 入内存队列self.mqs.push(request, -request.priority)

在上一篇文章时有说到&＃xff0c;调度器主要定义了 2 种队列&＃xff1a;基于磁盘队列、基于内存队列。

如果在实例化 Scheduler 时候传入 jobdir&＃xff0c;则使用磁盘队列&＃xff0c;否则使用内存队列&＃xff0c;默认使用内存队列。

指纹过滤

上面说到&＃xff0c;在请求入队之前&＃xff0c;首先会通过请求指纹过滤器检查请求是否重复&＃xff0c;也就是调用了过滤器的 request_seen&＃xff1a;

def request_seen(self, request):# 生成请求指纹fp &＃61; self.request_fingerprint(request)# 请求指纹如果在指纹集合中则认为重复if fp in self.fingerprints:return True# 不重复则记录此指纹self.fingerprints.add(fp)# 实例化如果有path则把指纹写入文件if self.file:self.file.write(fp &＃43; os.linesep)def request_fingerprint(self, request):# 调用utils.request的request_fingerprintreturn request_fingerprint(request)

utils.request 的 request_fingerprint 逻辑如下&＃xff1a;

def request_fingerprint(request, include_headers&＃61;None):"""生成请求指纹"""# 指纹生成是否包含headersif include_headers:include_headers &＃61; tuple(to_bytes(h.lower())for h in sorted(include_headers))cache &＃61; _fingerprint_cache.setdefault(request, {})if include_headers not in cache:# 使用sha1算法生成指纹fp &＃61; hashlib.sha1()fp.update(to_bytes(request.method))fp.update(to_bytes(canonicalize_url(request.url)))fp.update(request.body or b&＃39;&＃39;)if include_headers:for hdr in include_headers:if hdr in request.headers:fp.update(hdr)for v in request.headers.getlist(hdr):fp.update(v)cache[include_headers] &＃61; fp.hexdigest()return cache[include_headers]

这个过滤器先是通过 Request 对象生成一个请求指纹&＃xff0c;在这里使用 sha1 算法&＃xff0c;并记录到指纹集合&＃xff0c;每次请求入队前先到这里验证一下指纹集合&＃xff0c;如果已存在&＃xff0c;则认为请求重复&＃xff0c;则不会重复入队列。

不过如果我想不校验重复&＃xff0c;也想重复爬取怎么办&＃xff1f;看 enqueue_request 的第一行判断&＃xff0c;仅需将 Request 实例的 dont_filter 设置为 True 就可以重复抓取此请求&＃xff0c;非常灵活。

Scrapy 就是通过此逻辑实现重复请求的过滤&＃xff0c;默认情况下&＃xff0c;重复请求是不会进行重复抓取的。

下载请求

请求第一次进来后&＃xff0c;肯定是不重复的&＃xff0c;那么则会正常进入调度器队列。之后下一次调度&＃xff0c;再次调用 _next_request_from_scheduler 方法&＃xff0c;此时调用调度器的 next_request 方法&＃xff0c;就是从调度器队列中取出一个请求&＃xff0c;这次就要开始进行网络下载了&＃xff0c;也就是调用 _download&＃xff1a;

def _download(self, request, spider):# 下载请求slot &＃61; self.slotslot.add_request(request)def _on_success(response):# 成功回调结果必须是Request或Responseassert isinstance(response, (Response, Request))if isinstance(response, Response):# 如果下载后结果为Response 返回Responseresponse.request &＃61; requestlogkws &＃61; self.logformatter.crawled(request, response, spider)logger.log(*logformatter_adapter(logkws), extra&＃61;{&＃39;spider&＃39;: spider})self.signals.send_catch_log(signal&＃61;signals.response_received, \response&＃61;response, request&＃61;request, spider&＃61;spider)return responsedef _on_complete(_):# 此次下载完成后继续进行下一次调度slot.nextcall.schedule()return _# 调用Downloader进行下载dwld &＃61; self.downloader.fetch(request, spider)# 注册成功回调dwld.addCallbacks(_on_success)# 结束回调dwld.addBoth(_on_complete)return dwld

在进行网络下载时&＃xff0c;调用了 Downloader 的 fetch&＃xff1a;

def fetch(self, request, spider):def _deactivate(response):# 下载结束后删除此记录self.active.remove(request)return response# 下载前记录处理中的请求self.active.add(request)# 调用下载器中间件download 并注册下载成功的回调方法是self._enqueue_requestdfd &＃61; self.middleware.download(self._enqueue_request, request, spider)# 注册结束回调return dfd.addBoth(_deactivate)

这里调用下载器中间件的 download&＃xff0c;并注册下载成功的回调方法是 _enqueue_request&＃xff0c;来看下载方法&＃xff1a;

def download(self, download_func, request, spider):&＃64;defer.inlineCallbacksdef process_request(request):# 如果下载器中间件有定义process_request 则依次执行for method in self.methods[&＃39;process_request&＃39;]:response &＃61; yield method(request&＃61;request, spider&＃61;spider)assert response is None or isinstance(response, (Response, Request)), \&＃39;Middleware %s.process_request must return None, Response or Request, got %s&＃39; % \(six.get_method_self(method).__class__.__name__, response.__class__.__name__)# 如果下载器中间件有返回值直接返回此结果if response:defer.returnValue(response)# 如果下载器中间件没有返回值&＃xff0c;则执行注册进来的方法也就是Downloader的_enqueue_requestdefer.returnValue((yield download_func(request&＃61;request,spider&＃61;spider)))&＃64;defer.inlineCallbacksdef process_response(response):assert response is not None, &＃39;Received None in process_response&＃39;if isinstance(response, Request):defer.returnValue(response)# 如果下载器中间件有定义process_response 则依次执行for method in self.methods[&＃39;process_response&＃39;]:response &＃61; yield method(request&＃61;request, response&＃61;response,spider&＃61;spider)assert isinstance(response, (Response, Request)), \&＃39;Middleware %s.process_response must return Response or Request, got %s&＃39; % \(six.get_method_self(method).__class__.__name__, type(response))if isinstance(response, Request):defer.returnValue(response)defer.returnValue(response)&＃64;defer.inlineCallbacksdef process_exception(_failure):exception &＃61; _failure.value# 如果下载器中间件有定义process_exception 则依次执行for method in self.methods[&＃39;process_exception&＃39;]:response &＃61; yield method(request&＃61;request, exception&＃61;exception,spider&＃61;spider)assert response is None or isinstance(response, (Response, Request)), \&＃39;Middleware %s.process_exception must return None, Response or Request, got %s&＃39; % \(six.get_method_self(method).__class__.__name__, type(response))if response:defer.returnValue(response)defer.returnValue(_failure)# 注册执行、错误、回调方法deferred &＃61; mustbe_deferred(process_request, request)deferred.addErrback(process_exception)deferred.addCallback(process_response)return deferred

在下载过程中&＃xff0c;首先找到所有定义好的下载器中间件&＃xff0c;包括内置定义好的&＃xff0c;也可以自己扩展下载器中间件&＃xff0c;下载前先依次执行 process_request&＃xff0c;可对 Request 进行加工、处理、校验等操作&＃xff0c;然后发起真正的网络下载&＃xff0c;也就是第一个参数 download_func&＃xff0c;在这里是 Downloader 的 _enqueue_request 方法&＃xff1a;

下载成功后回调 Downloader的 _enqueue_request&＃xff1a;

def _enqueue_request(self, request, spider):# 加入下载请求队列key, slot &＃61; self._get_slot(request, spider)request.meta[&＃39;download_slot&＃39;] &＃61; keydef _deactivate(response):slot.active.remove(request)return responseslot.active.add(request)deferred &＃61; defer.Deferred().addBoth(_deactivate)# 下载队列slot.queue.append((request, deferred))# 处理下载队列self._process_queue(spider, slot)return deferreddef _process_queue(self, spider, slot):if slot.latercall and slot.latercall.active():return# 如果延迟下载参数有配置则延迟处理队列now &＃61; time()delay &＃61; slot.download_delay()if delay:penalty &＃61; delay - now &＃43; slot.lastseenif penalty > 0:slot.latercall &＃61; reactor.callLater(penalty, self._process_queue, spider, slot)return# 处理下载队列while slot.queue and slot.free_transfer_slots() > 0:slot.lastseen &＃61; now# 从下载队列中取出下载请求request, deferred &＃61; slot.queue.popleft()# 开始下载dfd &＃61; self._download(slot, request, spider)dfd.chainDeferred(deferred)# 延迟if delay:self._process_queue(spider, slot)breakdef _download(self, slot, request, spider):# 注册方法调用handlers的download_requestdfd &＃61; mustbe_deferred(self.handlers.download_request, request, spider)# 注册下载完成回调方法def _downloaded(response):self.signals.send_catch_log(signal&＃61;signals.response_downloaded,response&＃61;response,request&＃61;request,spider&＃61;spider)return responsedfd.addCallback(_downloaded)slot.transferring.add(request)def finish_transferring(_):slot.transferring.remove(request)# 下载完成后调用_process_queueself._process_queue(spider, slot)return _return dfd.addBoth(finish_transferring)

这里也维护了一个下载队列&＃xff0c;可根据配置达到延迟下载的要求。真正发起下载请求是调用了 self.handlers.download_request&＃xff1a;

def download_request(self, request, spider):# 获取请求的schemescheme &＃61; urlparse_cached(request).scheme# 根据scheeme获取下载处理器handler &＃61; self._get_handler(scheme)if not handler:raise NotSupported("Unsupported URL scheme &＃39;%s&＃39;: %s" %(scheme, self._notconfigured[scheme]))# 开始下载并返回结果return handler.download_request(request, spider)def _get_handler(self, scheme):# 根据scheme获取对应的下载处理器# 配置文件中定义好了http、https、ftp等资源的下载处理器if scheme in self._handlers:return self._handlers[scheme]if scheme in self._notconfigured:return Noneif scheme not in self._schemes:self._notconfigured[scheme] &＃61; &＃39;no handler available for that scheme&＃39;return Nonepath &＃61; self._schemes[scheme]try:# 实例化下载处理器dhcls &＃61; load_object(path)dh &＃61; dhcls(self._crawler.settings)except NotConfigured as ex:self._notconfigured[scheme] &＃61; str(ex)return Noneexcept Exception as ex:logger.error(&＃39;Loading "%(clspath)s" for scheme "%(scheme)s"&＃39;,{"clspath": path, "scheme": scheme},exc_info&＃61;True, extra&＃61;{&＃39;crawler&＃39;: self._crawler})self._notconfigured[scheme] &＃61; str(ex)return Noneelse:self._handlers[scheme] &＃61; dhreturn self._handlers[scheme]

下载前&＃xff0c;先通过解析 request 的 scheme 来获取对应的下载处理器&＃xff0c;默认配置文件中定义的下载处理器如下&＃xff1a;

DOWNLOAD_HANDLERS_BASE &＃61; {&＃39;file&＃39;: &＃39;scrapy.core.downloader.handlers.file.FileDownloadHandler&＃39;,&＃39;http&＃39;: &＃39;scrapy.core.downloader.handlers.http.HTTPDownloadHandler&＃39;,&＃39;https&＃39;: &＃39;scrapy.core.downloader.handlers.http.HTTPDownloadHandler&＃39;,&＃39;s3&＃39;: &＃39;scrapy.core.downloader.handlers.s3.S3DownloadHandler&＃39;,&＃39;ftp&＃39;: &＃39;scrapy.core.downloader.handlers.ftp.FTPDownloadHandler&＃39;, }

然后调用 download_request 方法&＃xff0c;完成网络下载&＃xff0c;这里不再详细讲解每个处理器的实现&＃xff0c;简单来说&＃xff0c;你可以把它想象成封装好的网络下载库&＃xff0c;输入URL&＃xff0c;它会给你输出下载结果&＃xff0c;这样方便理解。

在下载过程中&＃xff0c;如果发生异常情况&＃xff0c;则会依次调用下载器中间件的 process_exception 方法&＃xff0c;每个中间件只需定义自己的异常处理逻辑即可。

如果下载成功&＃xff0c;则会依次执行下载器中间件的 process_response 方法&＃xff0c;每个中间件可以进一步处理下载后的结果&＃xff0c;最终返回。

这里值得提一下&＃xff0c;process_request 方法是每个中间件顺序执行的&＃xff0c;而 process_response 和 process_exception 方法是每个中间件倒序执行的&＃xff0c;具体可看一下 DownaloderMiddlewareManager 的 _add_middleware 方法&＃xff0c;就可以明白是如何注册这个方法链的。

拿到最终的下载结果后&＃xff0c;再回到 ExecuteEngine 的 _next_request_from_scheduler 中&＃xff0c;会看到调用了 _handle_downloader_output&＃xff0c;也就是处理下载结果的逻辑&＃xff1a;

def _handle_downloader_output(self, response, request, spider):# 下载结果必须是Request、Response、Failure其一assert isinstance(response, (Request, Response, Failure)), response# 如果是Request 则再次调用crawl 执行Scheduler的入队逻辑if isinstance(response, Request):self.crawl(response, spider)return# 如果是Response或Failure 则调用scraper的enqueue_scrape进一步处理# 主要是和Spiders和Pipeline交互d &＃61; self.scraper.enqueue_scrape(response, request, spider)d.addErrback(lambda f: logger.error(&＃39;Error while enqueuing downloader output&＃39;,exc_info&＃61;failure_to_exc_info(f),extra&＃61;{&＃39;spider&＃39;: spider}))return d

拿到下载结果后&＃xff0c;主要分 2 个逻辑&＃xff1a;

如果返回的是 Request 实例&＃xff0c;则直接再次放入 Scheduler 请求队列
如果返回的是是 Response 或 Failure 实例&＃xff0c;则调用 Scraper 的 enqueue_scrape 方法&＃xff0c;做进一步处理

处理下载结果

请求入队逻辑不用再说&＃xff0c;前面已经讲过。现在主要看 Scraper 的 enqueue_scrape&＃xff0c;看Scraper 组件是如何处理后续逻辑的&＃xff1a;

def enqueue_scrape(self, response, request, spider):# 加入Scrape处理队列slot &＃61; self.slotdfd &＃61; slot.add_response_request(response, request)def finish_scraping(_):slot.finish_response(response, request)self._check_if_closing(spider, slot)self._scrape_next(spider, slot)return _dfd.addBoth(finish_scraping)dfd.addErrback(lambda f: logger.error(&＃39;Scraper bug processing %(request)s&＃39;,{&＃39;request&＃39;: request},exc_info&＃61;failure_to_exc_info(f),extra&＃61;{&＃39;spider&＃39;: spider}))self._scrape_next(spider, slot)return dfddef _scrape_next(self, spider, slot):while slot.queue:# 从Scraper队列中获取一个待处理的任务response, request, deferred &＃61; slot.next_response_request_deferred()self._scrape(response, request, spider).chainDeferred(deferred)def _scrape(self, response, request, spider):assert isinstance(response, (Response, Failure))# 调用_scrape2继续处理dfd &＃61; self._scrape2(response, request, spider)# 注册异常回调dfd.addErrback(self.handle_spider_error, request, response, spider)# 出口回调dfd.addCallback(self.handle_spider_output, request, response, spider)return dfddef _scrape2(self, request_result, request, spider):# 如果结果不是Failure实例则调用爬虫中间件管理器的scrape_responseif not isinstance(request_result, Failure):return self.spidermw.scrape_response(self.call_spider, request_result, request, spider)else:# 直接调用call_spiderdfd &＃61; self.call_spider(request_result, request, spider)return dfd.addErrback(self._log_download_errors, request_result, request, spider)

首先把请求和响应加入到 Scraper 的处理队列中&＃xff0c;然后从队列中获取到任务&＃xff0c;如果不是异常结果&＃xff0c;则调用爬虫中间件管理器的 scrape_response 方法&＃xff1a;

def scrape_response(self, scrape_func, response, request, spider):fname &＃61; lambda f:&＃39;%s.%s&＃39; % (six.get_method_self(f).__class__.__name__,six.get_method_function(f).__name__)def process_spider_input(response):# 执行一系列爬虫中间件的process_spider_inputfor method in self.methods[&＃39;process_spider_input&＃39;]:try:result &＃61; method(response&＃61;response, spider&＃61;spider)assert result is None, \&＃39;Middleware %s must returns None or &＃39; \&＃39;raise an exception, got %s &＃39; \% (fname(method), type(result))except:return scrape_func(Failure(), request, spider)# 执行完中间件的一系列process_spider_input方法后执行call_spiderreturn scrape_func(response, request, spider)def process_spider_exception(_failure):# 执行一系列爬虫中间件的process_spider_exceptionexception &＃61; _failure.valuefor method in self.methods[&＃39;process_spider_exception&＃39;]:result &＃61; method(response&＃61;response, exception&＃61;exception, spider&＃61;spider)assert result is None or _isiterable(result), \&＃39;Middleware %s must returns None, or an iterable object, got %s &＃39; % \(fname(method), type(result))if result is not None:return resultreturn _failuredef process_spider_output(result):# 执行一系列爬虫中间件的process_spider_outputfor method in self.methods[&＃39;process_spider_output&＃39;]:result &＃61; method(response&＃61;response, result&＃61;result, spider&＃61;spider)assert _isiterable(result), \&＃39;Middleware %s must returns an iterable object, got %s &＃39; % \(fname(method), type(result))return result# 执行process_spider_inputdfd &＃61; mustbe_deferred(process_spider_input, response)# 注册异常回调dfd.addErrback(process_spider_exception)# 注册出口回调dfd.addCallback(process_spider_output)return dfd

有没有感觉套路很熟悉&＃xff1f;与上面下载器中间件调用方式非常相似&＃xff0c;也调用一系列的前置方法&＃xff0c;再执行真正的处理逻辑&＃xff0c;最后执行一系列的后置方法。

回调爬虫

接下来看一下&＃xff0c;Scrapy 是如何执行我们写好的爬虫逻辑的&＃xff0c;也就是 call_spider 方法&＃xff0c;这里回调我们写好的爬虫类&＃xff1a;

def call_spider(self, result, request, spider):# 回调爬虫模块result.request &＃61; requestdfd &＃61; defer_result(result)# 注册回调方法取得request.callback 如果未定义则调用爬虫模块的parse方法dfd.addCallbacks(request.callback or spider.parse, request.errback)return dfd.addCallback(iterate_spider_output)

看到这里&＃xff0c;你应该更熟悉&＃xff0c;平时我们写的最多的爬虫代码&＃xff0c;parse 则是第一个回调方法。之后爬虫类拿到下载结果&＃xff0c;就可以定义下载后的 callback 方法&＃xff0c;也是在这里进行回调执行的。

处理输出

在与爬虫类交互完成之后&＃xff0c;Scraper 调用了 handle_spider_output 方法处理爬虫的输出结果&＃xff1a;

def handle_spider_output(self, result, request, response, spider):# 处理爬虫输出结果if not result:return defer_succeed(None)it &＃61; iter_errback(result, self.handle_spider_error, request, response, spider)# 注册_process_spidermw_outputdfd &＃61; parallel(it, self.concurrent_items,self._process_spidermw_output, request, response, spider)return dfddef _process_spidermw_output(self, output, request, response, spider):# 处理Spider模块返回的每一个Request/Itemif isinstance(output, Request):# 如果结果是Request 再次入Scheduler的请求队列self.crawler.engine.crawl(request&＃61;output, spider&＃61;spider)elif isinstance(output, (BaseItem, dict)):# 如果结果是BaseItem/dictself.slot.itemproc_size &＃43;&＃61; 1# 调用Pipeline的process_itemdfd &＃61; self.itemproc.process_item(output, spider)dfd.addBoth(self._itemproc_finished, output, response, spider)return dfdelif output is None:passelse:typename &＃61; type(output).__name__logger.error(&＃39;Spider must return Request, BaseItem, dict or None, &＃39;&＃39;got %(typename)r in %(request)s&＃39;,{&＃39;request&＃39;: request, &＃39;typename&＃39;: typename},extra&＃61;{&＃39;spider&＃39;: spider})

执行完我们自定义的解析逻辑后&＃xff0c;解析方法可返回新的 Request 或 BaseItem 实例。

如果是新的请求&＃xff0c;则再次通过 Scheduler 进入请求队列&＃xff0c;如果是 BaseItem 实例&＃xff0c;则调用 Pipeline 管理器&＃xff0c;依次执行 process_item。我们想输出结果时&＃xff0c;只需要定义 Pepeline 类&＃xff0c;然后重写这个方法就可以了。

ItemPipeManager 处理逻辑&＃xff1a;

class ItemPipelineManager(MiddlewareManager):component_name &＃61; &＃39;item pipeline&＃39;&＃64;classmethoddef _get_mwlist_from_settings(cls, settings):return build_component_list(settings.getwithbase(&＃39;ITEM_PIPELINES&＃39;))def _add_middleware(self, pipe):super(ItemPipelineManager, self)._add_middleware(pipe)if hasattr(pipe, &＃39;process_item&＃39;):self.methods[&＃39;process_item&＃39;].append(pipe.process_item)def process_item(self, item, spider):# 依次调用Pipeline的process_itemreturn self._process_chain(&＃39;process_item&＃39;, item, spider)

可以看到 ItemPipeManager 也是一个中间件&＃xff0c;和之前下载器中间件管理器和爬虫中间件管理器类似&＃xff0c;如果子类有定义 process_item&＃xff0c;则依次执行它。

执行完之后&＃xff0c;调用 _itemproc_finished&＃xff1a;

def _itemproc_finished(self, output, item, response, spider):self.slot.itemproc_size -&＃61; 1if isinstance(output, Failure):ex &＃61; output.value# 如果在Pipeline处理中抛DropItem异常忽略处理结果if isinstance(ex, DropItem):logkws &＃61; self.logformatter.dropped(item, ex, response, spider)logger.log(*logformatter_adapter(logkws), extra&＃61;{&＃39;spider&＃39;: spider})return self.signals.send_catch_log_deferred(signal&＃61;signals.item_dropped, item&＃61;item, response&＃61;response,spider&＃61;spider, exception&＃61;output.value)else:logger.error(&＃39;Error processing %(item)s&＃39;, {&＃39;item&＃39;: item},exc_info&＃61;failure_to_exc_info(output),extra&＃61;{&＃39;spider&＃39;: spider})else:logkws &＃61; self.logformatter.scraped(output, response, spider)logger.log(*logformatter_adapter(logkws), extra&＃61;{&＃39;spider&＃39;: spider})return self.signals.send_catch_log_deferred(signal&＃61;signals.item_scraped, item&＃61;output, response&＃61;response,spider&＃61;spider)

这里可以看到&＃xff0c;如果想在 Pipeline 中丢弃某个结果&＃xff0c;直接抛出 DropItem 异常即可&＃xff0c;Scrapy 会进行对应的处理。

到这里&＃xff0c;抓取结果会根据自定义的输出类&＃xff0c;然后输出到指定位置&＃xff0c;而新的 Request 则会再次进入请求队列&＃xff0c;等待引擎下一次调度&＃xff0c;也就是再次调用 ExecutionEngine 的 _next_request&＃xff0c;直至请求队列没有新的任务&＃xff0c;整个程序退出。

CrawlerSpider

以上&＃xff0c;基本上整个核心抓取流程就讲完了。

这里再简单说一下 CrawlerSpider 类&＃xff0c;我们平时用的也比较多&＃xff0c;它其实就是继承了 Spider 类&＃xff0c;然后重写了 parse 方法&＃xff08;这也是继承此类不要重写此方法的原因&＃xff09;&＃xff0c;并结合 Rule 规则类&＃xff0c;来完成 Request 的自动提取逻辑。

Scrapy 提供了这个类方便我们更快速地编写爬虫代码&＃xff0c;我们也可以基于此类进行再次封装&＃xff0c;让我们的爬虫代码写得更简单。

由此我们也可看出&＃xff0c;Scrapy 的每个模块的实现都非常纯粹&＃xff0c;每个组件都通过配置文件定义连接起来&＃xff0c;如果想要扩展或替换&＃xff0c;只需定义并实现自己的处理逻辑即可&＃xff0c;其他模块均不受任何影响&＃xff0c;所以我们也可以看到&＃xff0c;业界有非常多的 Scrapy 插件&＃xff0c;都是通过此机制来实现的。

总结

这篇文章的代码量较多&＃xff0c;也是 Scrapy 最为核心的抓取流程&＃xff0c;如果你能把这块逻辑搞清楚了&＃xff0c;那对 Scrapy 开发新的插件&＃xff0c;或者在它的基础上进行二次开发也非常简单了。

总结一下整个抓取流程&＃xff0c;还是用这两张图表示再清楚不过&＃xff1a;

Scrapy 整体给我的感觉是&＃xff0c;虽然它只是个单机版的爬虫框架&＃xff0c;但我们可以非常方便地编写插件&＃xff0c;或者自定义组件替换默认的功能&＃xff0c;从而定制化我们自己的爬虫&＃xff0c;最终可以实现一个功能强大的爬虫框架&＃xff0c;例如分布式、代理调度、并发控制、可视化、监控等功能&＃xff0c;它的灵活度非常高。

更多阅读

2020 年最佳流行 Python 库 Top 10

2020 Python中文社区热门文章 Top 10

5分钟快速掌握 Python 定时任务框架

特别推荐