深入理解 Scrapy 的 Pipeline
今天我们来深入学习 Scrapy 框架 Pipeline 的工作原理。这一次我们采取一种新的学习方式：先提出疑问，然后从源码中进行解答，直到最后我们彻底搞清楚 Pipeline 的工作流程。
1. 问题描述
这一小节我们将从源码的角度来分析 Pipeline 的工作过程。现在我先提出几个疑问：
- Scrapy 框架中使用 Pipeline 处理 Item 的代码在哪里？为什么我在 settings.py 中设置了 ITEM_PIPELINES 属性值，Scrapy 就能将其作为 Pipeline 去处理对应 Spider 生成的 Item 呢？
- 定义 Pipeline 的那四个方法来自哪里？为什么一定需要 process_item() 方法？
- 第12节中抓取起点月票榜小说时用到了图片管道，该管道的一个详细的处理流程是怎样的，即它如何实现图片下载？
带着这些疑问，我们来进入源码中寻找答案。
2. 源码解惑
2.1 Item Pipeline 的管理器类
还记得上一小节我们追踪 Spider 中间件的代码时，在 scrapy/core/scraper.py 中找到了 Spider 中间件处理 Spider 模块返回结果的方法，其代码内容如下：
# 源码位置：scrapy/core/scraper.py
# ...
class Scraper:
# ...
def _process_spidermw_output(self, output, request, response, spider):
"""Process each Request/Item (given in the output parameter) returned
from the given spider
"""
if isinstance(output, Request):
# 如果spider中间件返回的是Request，则继续调用引擎去处理请求
self.crawler.engine.crawl(request=output, spider=spider)
elif is_item(output):
# 如果spider中间件返回的是item，则调用self.itemproc对象的process_item()方法处理
self.slot.itemproc_size += 1
dfd = self.itemproc.process_item(output, spider)
dfd.addBoth(self._itemproc_finished, output, response, spider)
return dfd
elif output is None:
pass
else:
# 打印错误日志
# ...
从上面的代码我们知道，对于 Spider 中间件模块最后返回的 Item 类型数据会调用 self.itemproc 对象的 process_item() 方法处理，那么这个 self.itemproc 对象是什么呢？找到 Scraper 类的 __init__() 方法：
# 源码位置：scrapy/core/scraper.py
# ...
class Scraper:
def __init__(self, crawler):
# ...
itemproc_cls = load_object(crawler.settings['ITEM_PROCESSOR'])
self.itemproc = itemproc_cls.from_crawler(crawler)
# ...
# ...
来看默认的配置中关于 ITEM_PROCESSOR 的值，如下：
# 源码位置: scrapy/settings/default_settings.py
# ...
ITEM_PROCESSOR = 'scrapy.pipelines.ItemPipelineManager'
# ...
单看这个类的名称，又是一个某某管理器类，前面我们学了下载中间件管理类、Spider 中间件管理类，分别管理下载中间件类以及 Spider 中间件类，维护所属类方法的处理顺序。这里我们也是需要一个同样功能的管理类，来保证依次处理相应的 Item pipelines。我们进入该管理器类，阅读其实现代码：
# 源码位置：scrapy/
from scrapy.middleware import MiddlewareManager
from scrapy.utils.conf import build_component_list
from scrapy.utils.defer import deferred_f_from_coro_f
class ItemPipelineManager(MiddlewareManager):
component_name = 'item pipeline'
@classmethod
def _get_mwlist_from_settings(cls, settings):
return build_component_list(settings.getwithbase('ITEM_PIPELINES'))
def _add_middleware(self, pipe):
super(ItemPipelineManager, self)._add_middleware(pipe)
if hasattr(pipe, 'process_item'):
self.methods['process_item'].append(deferred_f_from_coro_f(pipe.process_item))
def process_item(self, item, spider):
return self._process_chain('process_item', item, spider)
同样，这个管理类直接就继承了前面的中间件管理器类，其代码量非常少，十分容易理解。
首先它和所有的中间件管理类一样从全局配置中获的对应管理的 pipelines，这个配置正是 ITEM_PIPELINES。其次，注意到这个 _add_middleware() 方法中有个调用父类的 _add_middleware() 方法，而父类中该方法的代码如下：
# 源码位置: scrapy/middleware.py
# ...
class MiddlewareManager:
# ...
def _add_middleware(self, mw):
if hasattr(mw, 'open_spider'):
self.methods['open_spider'].append(mw.open_spider)
if hasattr(mw, 'close_spider'):
self.methods['close_spider'].appendleft(mw.close_spider)
我们从而得知，在 pipeline 中会将 open_spider()、close_spider() 以及 process_item() 方法加入到对应的处理链中，且 MiddlewareManager 类中 from_crawler() 是一个类方法，因此对于继承该类的子类也同样会有该方法，也即具备了通过 Crawler 类对象实例化的能力。
2.2 Scrapy 框架内置的 Pipelines 分析
前面第12节中，我们在介绍 Scrapy 框架的管道内容时，使用了其内置的图片处理管道 (ImagesPipeline)，它对应的代码位置为：scrapy/pipelines/images.py。接下来，我们将分析其源码，看看如何实现图片下载的功能。
首先看看类继承关系：在 images.py 中定义的 ImagesPipeline 继承了 files.py 中定义的 FilesPipeline 类；而 FilesPipeline 类又继承至 media.py 中定义的 MediaPipeline 类。因此，我们先从分析基类开始，我们从管道的两个核心方法开始入手：
- 初始化方法：
__init__()； - Item 核心处理方法：
process_item()；
首先来看初始化的代码，如下：
# 源码位置：scrapy/pipelines/media.py
# ...
class MediaPipeline:
LOG_FAILED_RESULTS = True
class SpiderInfo:
def __init__(self, spider):
self.spider = spider
self.downloading = set()
self.downloaded = {}
self.waiting = defaultdict(list)
def __init__(self, download_func=None, settings=None):
self.download_func = download_func
if isinstance(settings, dict) or settings is None:
settings = Settings(settings)
resolve = functools.partial(self._key_for_pipe,
base_class_name="MediaPipeline",
settings=settings)
self.allow_redirects = settings.getbool(
resolve('MEDIA_ALLOW_REDIRECTS'), False
)
self._handle_statuses(self.allow_redirects)
def _handle_statuses(self, allow_redirects):
# 默认不允许重定向
self.handle_httpstatus_list = None
if allow_redirects:
# 当设置了allow_redirects时，会考虑处理存在3xx的下载地址
self.handle_httpstatus_list = SequenceExclude(range(300, 400))
def _key_for_pipe(self, key, base_class_name=None, settings=None):
"""
>>> MediaPipeline()._key_for_pipe("IMAGES")
'IMAGES'
>>> class MyPipe(MediaPipeline):
... pass
>>> MyPipe()._key_for_pipe("IMAGES", base_class_name="MediaPipeline")
'MYPIPE_IMAGES'
"""
class_name = self.__class__.__name__
formatted_key = "{}_{}".format(class_name.upper(), key)
if (
not base_class_name
or class_name == base_class_name
or settings and not settings.get(formatted_key)
):
return key
return formatted_key
# ...
上面的类中又定义了一个类：SpiderInfo ，这个类只是用来保存多个数据用的。此外，初始化方法中主要读取相关的配置，判断是否需要允许下载的 URL 重定向。该参数在 Scrapy 官方文档中的说明如下：
接下来是核心的处理 Item 的方法：
# 源码位置：scrapy/pipelines/media.py
# ...
class MediaPipeline:
# ...
def process_item(self, item, spider):
info = self.spiderinfo
# 从item中获取请求列表
requests = arg_to_iter(self.get_media_requests(item, info))
# 形成相关的处理链表
dlist = [self._process_request(r, info) for r in requests]
dfd = DeferredList(dlist, consumeErrors=1)
# 上述的处理全部完成后的回调
return dfd.addCallback(self.item_completed, item, info)
# ...
我们知道管道类中处理 Item 的核心方法是 process_item() ，上面的 process_item() 方法先调用对象的 get_media_requests() 方法从输入的 item 中获取相应的请求列表，然后在形成对应的请求列表，处理请求的方法为：_process_request()，最后所有的请求完成后会执行对象的 item_completed() 方法。
# 源码位置：scrapy/pipelines/media.py
# ...
class MediaPipeline:
# ...
def get_media_requests(self, item, info):
"""Returns the media requests to download"""
pass
# ...
这个 get_media_requests() 需要在后续的继承类中实现。接下来看处理下载请求的方法：
# 源码位置：scrapy/pipelines/media.py
# ...
class MediaPipeline:
# ...
def _process_request(self, request, info):
# 每个请求计算一个指纹，以保证后面不重复请求
fp = request_fingerprint(request)
# 请求回调
cb = request.callback or (lambda _: _)
# 请求错误回调
eb = request.errback
request.callback = None
request.errback = None
# 如果已经请求过了，直接取缓存的结果
if fp in info.downloaded:
return defer_result(info.downloaded[fp]).addCallbacks(cb, eb)
# Otherwise, wait for result
wad = Deferred().addCallbacks(cb, eb)
# 将请求的回调链加入对应的请求key中
info.waiting[fp].append(wad)
# 检查请求是否正在下载中，避免二次请求
if fp in info.downloading:
return wad
# 将请求加入正在下载的队列
info.downloading.add(fp)
# 创建Deferred对象，对应方法为self.media_to_download()
dfd = mustbe_deferred(self.media_to_download, request, info)
# 在self.media_to_download()方法处理完后回调self._check_media_to_download()方法
dfd.addCallback(self._check_media_to_download, request, info)
# 此外，再加入统一回调方法
dfd.addBoth(self._cache_result_and_execute_waiters, fp, info)
dfd.addErrback(lambda f: logger.error(
f.value, exc_info=failure_to_exc_info(f), extra={'spider': info.spider})
)
return dfd.addBoth(lambda _: wad) # it must return wad at last
# ...
上面请求的过程在注释中已详细说明，这里处理下载请求主要涉及的方法为：self.media_to_download() 以及 self._check_media_to_download()。我们继续查看该方法的代码：
# 源码位置：scrapy/pipelines/media.py
# ...
class MediaPipeline:
# ...
# Overridable Interface
def media_to_download(self, request, info):
"""Check request before starting download"""
pass
def _check_media_to_download(self, result, request, info):
if result is not None:
return result
if self.download_func:
# this ugly code was left only to support tests. TODO: remove
dfd = mustbe_deferred(self.download_func, request, info.spider)
dfd.addCallbacks(
callback=self.media_downloaded, callbackArgs=(request, info),
errback=self.media_failed, errbackArgs=(request, info))
else:
self._modify_media_request(request)
# 将请求发给引擎模块，调用download()方法下载网页
dfd = self.crawler.engine.download(request, info.spider)
dfd.addCallbacks(
callback=self.media_downloaded, callbackArgs=(request, info),
errback=self.media_failed, errbackArgs=(request, info))
return dfd
# ...
可以看到 media_to_download() 方法也是在继承类中需要重写的，而 _check_media_to_download() 方法则是核心处理下载文件或者图片的地方。该方法中首先判断是否有传入的 download_func() 方法用于下载网页，如果没有则调用引擎模块中的 download() 方法下载网页数据，成功后调用 media_downloaded() 方法，失败则调用 media_failed() 方法。最后我们来看下 self._cache_result_and_execute_waiters() 方法，其内容和逻辑比较简单，就是缓存请求的数据并将请求清除等待队列：
# 源码位置：scrapy/pipelines/media.py
# ...
class MediaPipeline:
# ...
def _cache_result_and_execute_waiters(self, result, fp, info):
if isinstance(result, Failure):
# minimize cached information for failure
result.cleanFailure()
result.frames = []
result.stack = None
context = getattr(result.value, '__context__', None)
if isinstance(context, StopIteration):
setattr(result.value, '__context__', None)
# 下载队列中移除该请求
info.downloading.remove(fp)
# 缓存下载请求结果
info.downloaded[fp] = result
# 移除等待队列中的该请求
for wad in info.waiting.pop(fp):
# 将原来请求的回调方法以及错误回调方法，加入回调处理链
defer_result(result).chainDeferred(wad)
此时，我们总结下 MediaPipeline 类的核心处理流程：
到此，MediaPipeline 类的核心方法我们已经研究完毕，接下来开始继续学习 MediaPipeline 这个类。注意到该类中并没有 process_item() 方法，因此它直接继承父类的 process_item() 方法。从 MediaPipeline 类中可知在 _check_media_to_download() 方法中会下载相应的媒体文件，成功后会回调 media_downloaded() 方法
# 源码位置：scrapy/pipelines/files.py
# ...
class FilesPipeline(MediaPipeline):
# ...
def media_downloaded(self, response, request, info):
referer = referer_str(request)
if response.status != 200:
# 打印告警信息，下载失败
# ...
raise FileException('download-error')
if not response.body:
# 打印告警信息，无下载内容
# ...
raise FileException('empty-content')
status = 'cached' if 'cached' in response.flags else 'downloaded'
# 打印debug信息
self.inc_stats(info.spider, status)
try:
# 设置下载文件路径
path = self.file_path(request, response=response, info=info)
# 将下载的内容保存成本地文件
checksum = self.file_downloaded(response, request, info)
except FileException as exc:
# 打印异常信息
# ...
raise
except Exception as exc:
# 打印异常信息
# ...
raise FileException(str(exc))
return {'url': request.url, 'path': path, 'checksum': checksum, 'status': status}
# ...
从上面的代码可知，在请求成功后，下载的内容保存在 response.body 中，上面的代码就是将该文件内容保存成磁盘上的文件：
# 源码位置：scrapy/pipelines/files.py
# ...
class FilesPipeline(MediaPipeline):
# ...
def file_downloaded(self, response, request, info):
# 生成文件保存路径
path = self.file_path(request, response=response, info=info)
# 获取字节流形式的下载内容
buf = BytesIO(response.body)
checksum = md5sum(buf)
buf.seek(0)
# 持久化保存
self.store.persist_file(path, buf, info)
# 返回文件的md5值
return checksum
上面的代码是不是已经够清楚了？最后文件内容是 buf，保存的方法是 self.store.persist_file(path, buf, info)，该方法是支持将下载内容保存成多种形式，比如保存到本地文件中、保存到 FTP 服务器上，甚至可以通过 S3 接口保存到云存储中。来看看保存成本地文件形式的代码，其实和我们平时写的文件操作一样，都是 open() 方法打开文件句柄，然后使用 wb 模式将内容写到文件中。
# 源码位置：scrapy/pipelines/files.py
# ...
class FSFilesStore:
# ...
def persist_file(self, path, buf, info, meta=None, headers=None):
absolute_path = self._get_filesystem_path(path)
self._mkdir(os.path.dirname(absolute_path), info)
# 保存文件
with open(absolute_path, 'wb') as f:
f.write(buf.getvalue())
# ...
最后对于 ImagesPipeline 类，其基本处理流程不变，只不过最后的保存方式和普通文件管道不一样，我们来看下面几个方法：
# 源码位置：scrapy/pipelines/images.py
# ...
class ImagesPipeline(FilesPipeline):
# ...
def file_downloaded(self, response, request, info):
return self.image_downloaded(response, request, info)
def image_downloaded(self, response, request, info):
checksum = None
for path, image, buf in self.get_images(response, request, info):
if checksum is None:
buf.seek(0)
checksum = md5sum(buf)
width, height = image.size
# 保存成图片形式
self.store.persist_file(
path, buf, info,
meta={'width': width, 'height': height},
headers={'Content-Type': 'image/jpeg'})
return checksum
def get_images(self, response, request, info):
path = self.file_path(request, response=response, info=info)
# 下载的图片内容主体
orig_image = Image.open(BytesIO(response.body))
width, height = orig_image.size
if width < self.min_width or height < self.min_height:
raise ImageException("Image too small (%dx%d < %dx%d)" %
(width, height, self.min_width, self.min_height))
image, buf = self.convert_image(orig_image)
yield path, image, buf
for thumb_id, size in self.thumbs.items():
thumb_path = self.thumb_path(request, thumb_id, response=response, info=info)
thumb_image, thumb_buf = self.convert_image(image, size)
yield thumb_path, thumb_image, thumb_buf
def convert_image(self, image, size=None):
# 图片转换格式
# ...
if size:
image = image.copy()
image.thumbnail(size, Image.ANTIALIAS)
buf = BytesIO()
image.save(buf, 'JPEG')
return image, buf
至于上面的代码细节限于篇幅就不再深究了，有兴趣的可以课后去深入学习，这里主要是使用了 Python 的一个专门用来处理图片的第三方模块：PIL。掌握了该模块的基本用法后，再看这些代码就一目了然了，都是非常常规和基础的代码。
好了，本小节的内容就到这里了。如果你能坚持看到这里，在回过头看看前面提出的问题，是否在心里都已经有了准确的答案？所有的疑问其实在看一遍源码之后便会豁然开朗，我们也能理解 Scrapy 中设置的参数的含义以及其作用，这些是我们后续深入定制化 Scrapy 框架的基础，一定要掌握。
3. 小结
本小节中我们先提出了 3 个问题，然后带着问题进入 Scrapy 的源码寻找答案，在最后完整看完 Pipeline 的工作代码后，在回过头来看原来的问题时，答案已经一目了然了。这种学习源码的方式也是非常有效的，带着问题去看代码。此外，我们没有深究代码细节，主要是根据架构图的数据导向来学习源码，课后也希望读者能继续深入这块的代码研究，提出问题，然后解答问题，最后完全掌握该模块。
- 还没有人评论，欢迎说说您的想法！
客服
