Scrapy源码解读

大数据技术架构

发布于 2023-03-08 15:05:37

6760

发布于 2023-03-08 15:05:37

Scrapy一个比较完整的爬虫框架，包含了爬取任务的调度、多个线程同时爬取（异步多线程，不用等一个请求完成后才开始另一个请求）、自动过滤重复的链接等功能。使用者通过定义比较简单的爬虫类（例如目标网址、爬取的具体页面元素、存储的格式字段、数据清理逻辑），剩余的就可以交给scrapy完成爬取工作。

Twisted

Twisted 是一个事件驱动的网络引擎。Twisted 是用于生成可扩展的跨平台网络服务器和客户端的引擎。在生产环境中以标准化方式轻松部署这些应用程序是此类平台获得广泛采用的重要组成部分。为此，Twisted 提供了一个应用程序基础架构：一种可重用和可配置的方式来部署 Twisted 应用程序。它允许程序员通过将应用程序挂接到现有工具中来避免样板代码，以自定义其运行方式，包括守护程序、日志记录、使用自定义反应器、分析代码等。

事件驱动event-driven的程序，在单个控制线程中交错执行三个任务。当在执行 I/O 或其他成本高昂的操作时，会注册一个callback回调函数，然后在 I/O 完成时继续执行程序。回调函数描述事件完成后如何处理事件。Event loop事件循环轮询poll，并在事件发生时将他们分发给回调函数。这样的方式，就允许程序在不使用多线程的情况下持续执行（协程的概念）。

例如一个网络请求，就是一个耗时等待操作，在请求网页之后需要等待页面响应并返回结果。耗时等待操作一般都是1O操作，例如文件读取、网络请求等。协程在处理这种操作时是有很大优势的，当遇到需要等待时，程序暂时挂起，转而执行其他操作，从而避免因一直等待一个程序而耗费过多的时间。

总之，Twisted 和 Asyncio 类，都是支持协程的，前者比后者出现的早，其核心都是事件循环。当程序执行到某个耗时的 IO 操作时，程序的执行权限会被退回给事件循环，事件循环会检测其它准备就绪的协程，然后将执行权限交给它，当之前的协程 IO 操作完毕后，事件循环会将执行权限转给它，继续后面的操作。这样就实现在单线程内实现并发，只是比多线程更轻量。事件循环在 Asyncio 中被叫做 event_loop，在 Twisted 中叫做 reactor。

Twisted 的核心是reactor event loop。reactor反应器知道网络、文件系统和计时器事件。它等待并解复用这些事件，并将它们调度到等待的事件处理程序。

A transport传输表示通过网络通信的两个终结点之间的连接。传输描述连接详细信息：例如，此连接是面向流的（如 TCP）还是面向数据报文的，如 UDP、TCP、UDP、Unix 套接字和串行端口等。

Protocols协议描述如何异步处理网络事件。Twisted 维护了许多流行应用程序协议的实现，包括 HTTP、Telnet、DNS 和IMAP。

Deferreds延迟有一对回调链，一个用于成功（回调），一个用于错误（错误）。延迟从两个空链开始。将回调和错误对添加到延迟对象，定义每个事件成功和失败情况下对应的操作。

Python生成器是一个“可重启的函数”,它是在函数体中用 yield 语句创建的. 这样做可以使这个函数变成一个“生成器函数”,它返回一个”iterator“可以用来以一系列步骤运行这个函数. 每个迭代循环都会重启这个函数，继续执行到下一个 yield 语句。这与异步系统中的回调工作方式非常类似. 我们可以把 while 循环视作 reactor, 把生成器视作一系列由 yield 语句分隔的回调函数. 生成器总是在每个 yield 语句后暂停直到被显示的重启.因而我们可以延迟它的重启直到 deferred 被激发, 届时我们会使用send 方法发送值(如果 deferred 成功)或者抛出异常(如果 deferred 失败)，这就使我们的生成器成为一个真正的异步回调序列,这正是 twisted.internet.defer 中 inlineCallbacks 函数背后的概念.

断点跟踪

Scrapy很多命令是通过cmd来完成，这个运行入口的原理参考 https://zhuanlan.zhihu.com/p/272370864，其本质是调用了scrapy包下面的__main__是入口函数。为了能够做断点调试看源码，可以在pycharm里面的run->edit configuration，直接输入cmd的命令，例如crawl quotes。新建一个文件，输入

from scrapy.cmdline import execute if __name__ == '__main__': execute()

断点run该文件。

源码解读

核心概念：

Engjne: 引擎是整个框架的核心，可理解为整个框架的中央处理器，把其他几个核心部件整合在一起，整体负责数据的流转和逻辑的处理。
Item:它是一个抽象的数据结构，定义爬取结果的数据结构，每个Item是—个类，类里面定义了爬取结果的数据字段，可以理解为它用来规定爬取数据的存储格式。
Scheduler:调度器，接受Engine发过来的Request并将其加入队列中，同时也可以将Request发回给Engine供Downloader执行下载，主要是维护request的队列逻辑，例如先进先出、先进后出、优先级进出等
Spiders：蜘蛛，每个Spider定义站点的爬取逻辑和页面的解析规则，主要负责解析响应并生成Item，产生新的请求再发给Engine进行处理。网站的链接、抓取逻辑、解析逻辑都在spider类中定义。
Downloader: 下载器，即完成“向服务器发送请求，然后拿到响应的过程’得到的响应会再发送给Engine处理
ItemPipelines:项目管道，主要负责处理由Spider从页面中抽取的Item，做一些数据清洗、验证和存储等工作，比如将Item的字段清洗后存储到数据库
DownloaderMiddlewares:下载器中间件，位于Engine和Downloader之间的Hook框架，负责实现Downloader和Engjne之间的请求和响应的处理过程。Request被Engine发送给Downloader执行下载之前，Downloader Middleware可以对Request进行修改(process_request)。Downloader执行Request后生成Response，在Response被Engine发送给Spider之前，即Resposne被Spider解析之前，它可以对Response进行修改(process_response)。修改User-Agent、处理重定向、设置代理、失败重试、设置Cookie等动态渲染、反爬处理功能都可以借助它来实现。
SpiderMiddlewares:蜘蛛中间件，它是位于Englne和Spiders之间的Hook框架，负责实现Spiders和Englne之间的Item请求和响应的处理过程。Spider Middleware可以用来处理输入给Spider的Response和Spider输出的Item以及Request，比如过滤一些不需要的request和response。
Extension：扩展件，可以添加和扩展一些自定义的功能。利用Extension可以注册一些处理方法并监听Scrapy运行过程中的信号(利用crawler的signals对象将Scrapy的各个信号和已经定义的处理方法关联起来)，发生某个事件时执行自定义的方法。例如LogStats用于记录一些基本的爬取信息，比如爬取的页面数量、提取的Item数量等。

scrapy的工作流程

用户定义spider，包含目标网址等
Scrapy Engine(核心引擎)，获得目标网址，同步给Scheduler(调度器，负责管理任务、过滤任务、输出任务、存储、去重任务都在此控制)。Engine按照schedule给的任务，持续发送网页浏览请求，爬取信息。
Scheduler按照事先获得的网址，陆续把爬取任务分配给Engine。
Engine发送请求给Downloader（下载器，负责在网络上下载数据，输入待下载的 URL，输出下载结果）,如果有配置Downloader middlewares，则接着调用中间件
Downloader完成任务后，在发给engine，Engine进一步发给spider对结果进行处理，然后经过Spider Middleware的处理。
Spider处理完以后，Engine发送结果给item pipeline（负责输出结构化数据，可自定义格式和输出的位置）。Spider发送下一个请求request给engine，engine再发给scheduler。

详细过程

在前面的文件中，from scrapy.cmdline import execute

execute()函数会执行如下步骤：

获得项目的配置信息：调用get_project_settings，方法closest_scrapy_cfg()定位优先级最高的.cfg文件。ConfigParser()是一个以正则方式读取cfg里的配置信息的类（读取模板变量）。Settings()是一个类似字典的类，加载scrapy包下默认的setting（site-packages/scrapy/settings/default_settings.py），以及项目文件夹下的setting.py获得爬虫具体的配置信息。
inside_project()利用是否能成功setting.py来判断，当前工作路径是否在项目内部
使用iter_modules动态加载scrapy.commands下的所有类，从scrapy.commands下加载各个cmd的python文件，每个文件都是一个类对象，分别对应着一种cmd命令。当前面输入的是crawl quotes的时候，就会调用crawl文件中定义的对象。
_run_print_help(parser, cmd.process_options, args, opts)是一个尝试运行命令，如果有报错会做出提示，退出运行。
cmd.crawler_process = CrawlerProcess(settings)，这是管理多个spider同时异步运行的类。初始化该类的时候，会加载项目文件夹里面的spider，加载的方法会根据setting里面设置的加载类（这个方法很不错，可以动态的通过设置setting来改变需要使用的类），如果自定义加载类，需要遵循scrapy.interfaces.ISpiderLoader的规范(使用zope.verifyClass来判断目标类是否包含所需的接口)。Spider loader初始化时，会加载项目的spider(通过vars()方法，遍历目标文件的属性和属性值的字典对象)。
_run_print_help(parser, _run_command, cmd, args, opts)开始执行Command的run函数，启动多个线程的爬取任务。这是一个异步函数，里面会对所有核心组件进行实例化，等到后面调用self.crawler_process.start()，才真正开始启动reactor事件循环，标志着所有爬虫正式运行。如果没有手动结束，会等待所有爬虫全部爬取完成后才结束。
在上面的函数内，_create_crawler根据setting加载自定义的spider，封装成crawler类，可以理解成专门管理爬虫运行的类。在这个类里包含SignalManager(接受信号以后进行回调)， ExtensionManager（各种中间件或者插件，比如对内存使用率数据的收集）等管理和监控爬虫的功能。SignalManager的运行机制是，使用信号分发器dispatcher.connect()，来设置信号和信号触发函数，当捕获到信号时执行一个函数。
Crawler类中的crawl使用@defer.inlineCallbacks来修饰，意思是这是一个延迟任务（异步任务），内部会通过yield语法来实现多个回调函数。首先，实例化之前用户定义好的spider，实例化execution engine（针对一个具体爬虫工作的管理功能）。通过start_requests = iter(self.spider.start_requests())获得在spider里定义的目标地址，而后调用engine.open_spider，里面会有如下操作：
实例化一个scheduler类（任务调度功能）
start_requests=yield self.scraper.spidermw.process_start_requests(start_requests, spider)的右边会使用spider中间件进行request处理（异步），再传给start_request，这里就可以通过自定义spidermiddleware来实现一些request的过滤操作。
初始化slot，可以理解为资源槽位。一个engine只允许有一个slot（我理解成是一个爬虫任务）。其一个属性Heartbeat心跳，按照一定时间频率启动某任务，本质上是调用twisted库的LoopingCall
yield self.scraper.open_spider(spider)，定义一个slot，该slot跟前面的slot不一样，会定义max_active_size用来限制最大同时爬取的任务（注意这里指的是异步任务，并不是真正的多线程）。比如在setting里面设置CONCURRENT REQUESTS =6我们将并发量修改为了6，这样在爬取过程中就会同时使用Chrome渲染6个页面了。如果电脑性能比较不错的话，可以将这个数字调得更大一些。调用itemproc.open_spider，完成对数据库连接的配置工作。scrapy基于twisted异步IO框架，大部分操作都是单线程的，downloader是可以多线程的(REACTOR_THREADPOOL_MAXSIZE配置改变启动的线程数，底层是通过reactor.getThreadPool来开辟线程池来完成)。
slot.nextcall.schedule，其中的nextcall实际上是CallLaterOnce(self._next_request)，调用_next_request
heartbeat.start 每5分钟调用一次_next_request
yield defer.maybeDeferred(self.engine.start)，启动执行引擎，记录爬虫开始时间。此时仍然并未真正开始爬取，仍然是CrawlerProcess.start()之前的预处理步骤。只有crawler_process.start()才真正开始爬取任务。
nextcall = CallLaterOnce(self._next_request)，这个_next_request里面实现了如下操作：
使用while not self._needs_backout()来控制爬取速度，如果没有获得新的地址或者资源已用满，就等待
如果是起始的链接，使用scheduler.enqueue_request将request放入队列
_next_request_from_scheduler会启动_handle_downloader_output，里面会调用enqueue_scrape，从而进一步调用callback = result.request.callback or spider._parse，也就是之前用户定义的parse规则，完成页面的自定义解析，获得数据，而后回调handle_spider_output，进一步完成中间件的数据处理（数据清洗、存储等）。如果有新的链接，就发送请求，通过dwld.addBoth(_on_complete)完成回调。在Scrapy中Request对象实际上指的就是scrapy.http.Request的一个实例，包含了HTTP请求的基本信息，从而进一步由Engine交给Downloader进行处理执行，返回一个Response对象。