scrapy-redis的官方文档写的比较简洁，没有提及其运行原理，所以如果想全面的理解分布式爬虫的运行原理，还是得看scrapy-redis的源代码才行。
官方站点：https://github.com/rolando/scrapy-redis
scrapy-redis工程的主体还是是redis和scrapy两个库，工程本身实现的东西不是很多，这个工程就像胶水一样，把这两个插件粘结了起来。下面我们来看看，scrapy-redis的每一个源代码文件都实现了什么功能，最后如何实现分布式的爬虫系统。
一、scrapy-redis架构原理
Scrapy-redis提供了下面四种组件（components）：
四种组件意味着这四个模块都要做相应的修改
- Scheduler（调度器）
- Duplication Filter（重复过滤）
- Item Pipeline（管道）
- Base Spider（继承类）
下面分别介绍四个组件：
1. Scheduler（调度器）：
Scrapy改造了python本来的collection.deque(双向队列)形成了自己的Scrapy queue(https://github.com/scrapy/queuelib/blob/master/queuelib/queue.py))，但是Scrapy多个spider不能共享待爬取队列Scrapy queue， 即Scrapy本身不支持爬虫分布式，scrapy-redis 的解决是把这个Scrapy queue换成redis数据库（也是指redis队列），从同一个redis-server存放要爬取的request，便能让多个spider去同一个数据库里读取。
Scrapy中跟“待爬队列”直接相关的就是调度器Scheduler，它负责对新的request进行入列操作（加入Scrapy queue），取出下一个要爬取的request（从Scrapy queue中取出）等操作。它把待爬队列按照优先级建立了一个字典结构，比如：
{
优先级0 : 队列0
优先级1 : 队列1
优先级2 : 队列2
}
然后根据request中的优先级，来决定该入哪个队列，出列时则按优先级较小的优先出列。为了管理这个比较高级的队列字典，Scheduler需要提供一系列的方法。但是原来的Scheduler已经无法使用，所以使用Scrapy-redis的scheduler组件。
2. Duplication Filter(过滤工具)：
Scrapy中用集合实现这个request去重功能，Scrapy中把已经发送的request指纹放入到一个集合中，把下一个request的指纹拿到集合中比对，如果该指纹存在于集合中，说明这个request发送过了，如果没有则继续操作。这个核心的判重功能是这样实现的：
def request_seen(self, request):
# self.request_figerprints 就是一个指纹集合
fp = self.request_fingerprint(request)
# 这就是判重的核心操作
if fp in self.fingerprints:
return True
self.fingerprints.add(fp)
if self.file:
self.file.write(fp + os.linesep)
在scrapy-redis中去重是由Duplication Filter组件来实现的，它通过redis的set 不重复的特性，巧妙的实现了Duplication Filter去重。scrapy-redis调度器从引擎接受request，将request的指纹存redis的set检查是否重复，并将不重复的request push写redis的 request queue。
引擎请求request(Spider发出的）时，调度器从redis的request queue队列里根据优先级pop 出个request 返回给引擎，引擎将此request发给spider处理。
3. Item Pipeline（管道）：
引擎将(Spider返回的)爬取到的Item给Item Pipeline，scrapy-redis 的Item Pipeline将爬取到的 Item 存redis的 items queue。
修改过Item Pipeline可以很方便的根据 key 从 items queue 提取item，从而实现items processes集群。
4. Base Spider：
不在使用scrapy原有的Spider类，重写的RedisSpider继承了Spider和RedisMixin这两个类，RedisMixin是用来从redis读取url的类。
当我们生成一个Spider继承RedisSpider时，调用setup_redis函数，这个函数会去连接redis数据库，然后会设置signals(信号)：
- 一个是当spider空闲时候的signal，会调用spider_idle函数，这个函数调用schedule_next_request函数，保证spider是一直活着的状态，并且抛出DontCloseSpider异常。
- 一个是当抓到一个item时的signal，会调用item_scraped函数，这个函数会调用schedule_next_request函数，获取下一个request。
Scrapy-redis框架执行过程总结：
最后总结一下scrapy-redis的总体思路：这套组件通过重写scheduler和 spider类，实现了调度、spider启动和redis的交互。
实现新的dupefilter和queue类，达到了判重和调度容器和redis 的交互，因为每个主机上的爬虫进程都访问同一个redis数据库，所以调度和判重都统一进行统一管理，达到了分布式爬虫的目的。
当spider被初始化时，同时会初始化一个对应的scheduler对象，这个调度器对象通过读取settings，配置好自己的调度容器queue和判重工具dupefilter。
每当一个spider产出一个request的时候，scrapy引擎会把这个reuqest递交给这个spider对应的scheduler对象进行调度，scheduler对象通过访问redis对request进行判重，如果不重复就把他添加进redis中的调度器队列里。当调度条件满足时，scheduler对象就从redis的调度器队列中取出一个request发送给spider，让他爬取。
当spider爬取的所有暂时可用url之后，scheduler发现这个spider对应的redis的调度器队列空了，于是触发信号spider_idle，spider收到这个信号之后，直接连接redis读取start_urls池，拿取新的一批url入口，然后再次重复上边的工作。
二、源码解析
下面的源码的注释基本都有，我就重要的代码进行解释
1.connection.py
这个文件是用于连接redis的文件，用到比较多，也是最重要的文件
import six
from scrapy.utils.misc import load_object
from . import defaults
# Shortcut maps 'setting name' -> 'parmater name'.
#关系
SETTINGS_PARAMS_MAP = {
'REDIS_URL': 'url',
'REDIS_HOST': 'host',
'REDIS_PORT': 'port',
'REDIS_ENCODING': 'encoding',
}
#获取一个redis连接实例
#生成连接redis的参数
def get_redis_from_settings(settings):
"""Returns a redis client instance from given Scrapy settings object.
This function uses ``get_client`` to instantiate the client and uses
``defaults.REDIS_PARAMS`` global as defaults values for the parameters. You
can override them using the ``REDIS_PARAMS`` setting.
Parameters
----------
settings : Settings
A scrapy settings object. See the supported settings below.
Returns
-------
server
Redis client instance.
Other Parameters
----------------
REDIS_URL : str, optional
Server connection URL.
REDIS_HOST : str, optional
Server host.
REDIS_PORT : str, optional
Server port.
REDIS_ENCODING : str, optional
Data encoding.
REDIS_PARAMS : dict, optional
Additional client parameters.
"""
#浅拷贝 是为了防止params的改变，会导致默认SETTINGS_PARAMS被改变
params = defaults.REDIS_PARAMS.copy()
#将设置中的参数更新进入params中
params.update(settings.getdict('REDIS_PARAMS'))
# XXX: Deprecate REDIS_* settings.
#遍历映射表，获取指定的参数
for source, dest in SETTINGS_PARAMS_MAP.items():
#优先使用设置中的参数
val = settings.get(source)
#如果设置中没有进行设置，则params不更新
if val:
params[dest] = val
# Allow ``redis_cls`` to be a path to a class.
if isinstance(params.get('redis_cls'), six.string_types):
params['redis_cls'] = load_object(params['redis_cls'])
return get_redis(**params)
# Backwards compatible alias.
from_settings = get_redis_from_settings
def get_redis(**kwargs):
"""Returns a redis client instance.
Parameters
----------
redis_cls : class, optional
Defaults to ``redis.StrictRedis``.
url : str, optional
If given, ``redis_cls.from_url`` is used to instantiate the class.
**kwargs
Extra parameters to be passed to the ``redis_cls`` class.
Returns
-------
server
Redis client instance.
"""
#没有redis_cls 则用默认的redis连接
redis_cls = kwargs.pop('redis_cls', defaults.REDIS_CLS)
#判断kwarg有没有url
url = kwargs.pop('url', None)
if url:
return redis_cls.from_url(url, **kwargs)
else:
#走这里
return redis_cls(**kwargs)
Connection提供了一个很重要的函数，from_settings = get_redis_from_settings这个函数引入defualt.py文件，定义了我们访问过的指纹。pipline，queue，schedule文件都会调用。
2.defaults.py
主要存放默认的参数
import redis
# For standalone use.
#去重的键名key
DUPEFILTER_KEY = 'dupefilter:%(timestamp)s'
#定义的存储items的键名，spiders是爬虫的名称
PIPELINE_KEY = '%(spider)s:items'
#redis连接对象，是用于连接redis
REDIS_CLS = redis.StrictRedis
#字符集编码
REDIS_ENCODING = 'utf-8'
# Sane connection defaults.
# redis的连接的参数
REDIS_PARAMS = {
'socket_timeout': 30,
'socket_connect_timeout': 30,
'retry_on_timeout': True,
'encoding': REDIS_ENCODING,
}
# 队列的变量名，用于存储爬取的url队列
SCHEDULER_QUEUE_KEY = '%(spider)s:requests'
# 优先级队列，用于规定队列的进出方式
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.PriorityQueue'
# 用于去重的key，给request加指纹存储的地方
SCHEDULER_DUPEFILTER_KEY = '%(spider)s:dupefilter'
# 用于生成指纹的类
SCHEDULER_DUPEFILTER_CLASS = 'scrapy_redis.dupefilter.RFPDupeFilter'
# 起始url对应的key
START_URLS_KEY = '%(name)s:start_urls'
# 起始url的类型
START_URLS_AS_SET = False
3. dupefilter.py
scrapy的去重是利用集合来实现的，而在scrapy分布式的去重就需要利用共享的集合，那么这里使用的就是redis中集合数据结构。
import logging
import time
from scrapy.dupefilters import BaseDupeFilter
from scrapy.utils.request import request_fingerprint
from . import defaults
from .connection import get_redis_from_settings
logger = logging.getLogger(__name__)
# TODO: Rename class to RedisDupeFilter.
class RFPDupeFilter(BaseDupeFilter):
"""
Redis-based request duplicates filter.
This class can also be used with default Scrapy's scheduler.
"""
logger = logger
def __init__(self, server, key, debug=False):
"""
Initialize the duplicates filter.
Parameters#参数
----------
#server:redis的连接实例
server : redis.StrictRedis
The redis server instance.
key : str
Redis key Where to store fingerprints. 存储requests指纹的地方
debug : bool, optional
Whether to log filtered requests. 是否记录过滤的requests
"""
self.server = server
self.key = key
self.debug = debug
self.logdupes = True
#类方法传递的当前方法
@classmethod
def from_settings(cls, settings):
"""Returns an instance from given settings.
This uses by default the key ``dupefilter:<timestamp>``. When using the
``scrapy_redis.scheduler.Scheduler`` class, this method is not used as
it needs to pass the spider name in the key.
Parameters
----------
settings : scrapy.settings.Settings
Returns
-------
RFPDupeFilter
A RFPDupeFilter instance.
"""
#获取redis的连接实例
server = get_redis_from_settings(settings)
# XXX: This creates one-time key. needed to support to use this
# class as standalone dupefilter with scrapy's default scheduler
# if scrapy passes spider on open() method this wouldn't be needed
# TODO: Use SCRAPY_JOB env as default and fallback to timestamp.
#生成存储指纹的key
key = defaults.DUPEFILTER_KEY % {'timestamp': int(time.time())}
#使用默认值Flase
debug = settings.getbool('DUPEFILTER_DEBUG')
#传给当前类，并把参数传给init函数
return cls(server, key=key, debug=debug)
@classmethod
def from_crawler(cls, crawler):
"""Returns instance from crawler.
Parameters
----------
crawler : scrapy.crawler.Crawler
Returns
-------
RFPDupeFilter
Instance of RFPDupeFilter.
"""
return cls.from_settings(crawler.settings)
def request_seen(self, request):
"""Returns True if request was already seen.
Parameters
----------
request : scrapy.http.Request
Returns
-------
bool
"""
#s生成一个指纹
fp = self.request_fingerprint(request)
# This returns the number of values added, zero if already exists.
#将指纹加入redis 指纹是一个集合类型
# self.server redis连接实例
#self.key 是存储指纹的key fp指纹
#self.key 已经存在返回0，不存在则返回1
added = self.server.sadd(self.key, fp)
#当added为0，说明指纹已经存在，返回true。否则返回False
return added == 0
def request_fingerprint(self, request):
"""Returns a fingerprint for a given request.
Parameters
----------
request : scrapy.http.Request
Returns
-------
str
"""
return request_fingerprint(request)
@classmethod
def from_spider(cls, spider):
settings = spider.settings
server = get_redis_from_settings(settings)
dupefilter_key = settings.get("SCHEDULER_DUPEFILTER_KEY", defaults.SCHEDULER_DUPEFILTER_KEY)
key = dupefilter_key % {'spider': spider.name}
debug = settings.getbool('DUPEFILTER_DEBUG')
return cls(server, key=key, debug=debug)
#当爬虫结束时，清空指纹
def close(self, reason=''):
"""Delete data on close. Called by Scrapy's scheduler.
Parameters
----------
reason : str, optional
"""
self.clear()
def clear(self):
"""Clears fingerprints data."""
self.server.delete(self.key)
def log(self, request, spider):
"""Logs given request.
Parameters
----------
request : scrapy.http.Request
spider : scrapy.spiders.Spider
"""
if self.debug:
msg = "Filtered duplicate request: %(request)s"
self.logger.debug(msg, {'request': request}, extra={'spider': spider})
elif self.logdupes:
msg = ("Filtered duplicate request %(request)s"
" - no more duplicates will be shown"
" (see DUPEFILTER_DEBUG to show all duplicates)")
self.logger.debug(msg, {'request': request}, extra={'spider': spider})
self.logdupes = False
request_seen（）方法直接换成了数据库的存储方式，鉴别重复的方式，还是使用指纹，指纹同样，是依靠request_fingerprint()方法俩获取的。获取指纹之后就直接向集合添加指纹，如果添加成功，说明这个指纹原本不存在于集合中，返回1。代码中最后的返回结果是判断添加结果是否为0，如果刚才返回为1，那么这个判定结果是false，也就是不重复，否则判定为重复。
4. picklecompat.py
这里实现了loads和dumps两个函数，其实就是实现了一个序列化器。
因为redis数据库不能存储复杂对象（key部分只能是字符串，value部分只能是字符串，字符串列表，字符串集合和hash），所以我们存啥都要先串行化成文本才行。
这里使用的就是python的pickle模块，一个兼容py2和py3的串行化工具。这个serializer主要用于一会的scheduler存reuqest对象。
"""A pickle wrapper module with protocol=-1 by default."""
try:
import cPickle as pickle # PY2
except ImportError:
import pickle#PY3用的包
#反序列化就是将字符串转换为json数据
def loads(s):
return pickle.loads(s)
#序列化就是将json数据转换为字符串
def dumps(obj):
return pickle.dumps(obj, protocol=-1)
5. piplines.py
用于处理爬虫爬取的数据将数据序列化放到redis中
from scrapy.utils.misc import load_object
from scrapy.utils.serialize import ScrapyJSONEncoder
from twisted.internet.threads import deferToThread
from . import connection, defaults
#序列化的字符串
default_serialize = ScrapyJSONEncoder().encode
class RedisPipeline(object):
"""Pushes serialized item into a redis list/queue
Settings
--------
REDIS_ITEMS_KEY : str
Redis key where to store items.
REDIS_ITEMS_SERIALIZER : str
Object path to serializer function.
"""
def __init__(self, server,
key=defaults.PIPELINE_KEY,
serialize_func=default_serialize):
"""Initialize pipeline.
Parameters
----------
server : StrictRedis
Redis client instance.
key : str
Redis key where to store items.
serialize_func : callable
Items serializer function.
"""
self.server = server
self.key = key
self.serialize = serialize_func
#将类本身传入函数
#用来生成参数和redis的连接实例
@classmethod
def from_settings(cls, settings):
#from_settings = get_redis_from_settings
#生成redis连接实例
params = {
'server': connection.from_settings(settings),
}
#如果设置中有item_key，我们就用设置中的
if settings.get('REDIS_ITEMS_KEY'):
params['key'] = settings['REDIS_ITEMS_KEY']
#如果设置中有序列化的函数，则优先使用设置中的
if settings.get('REDIS_ITEMS_SERIALIZER'):
params['serialize_func'] = load_object(
settings['REDIS_ITEMS_SERIALIZER']
)
#将参数返回当前类
return cls(**params)
@classmethod
def from_crawler(cls, crawler):
return cls.from_settings(crawler.settings)
#将item传递过来，自动触发这个函数，process_item
def process_item(self, item, spider):
#创建一个线程，用于存储item，也就是说上一个item还没有存储完，下一个item就可以存储
return deferToThread(self._process_item, item, spider)
#实现存储函数
def _process_item(self, item, spider):
#生成item_key
key = self.item_key(item, spider)
#使用默认的序列化函数，将item序列化为字符串
data = self.serialize(item)
#self.server是redis的连接实例
self.server.rpush(key, data)
return item
#用于存储item
def item_key(self, item, spider):
"""Returns redis key based on given spider.
Override this function to use a different key depending on the item
and/or spider.
"""
# self.key='%(spider)s:items'=%(spider.name)s:items'
return self.key % {'spider': spider.name} # 格式化字符串
6.queue.py
爬取队列，有三个队列实现，首先它实现了一个父类base，提供一些基本方法与属性
from scrapy.utils.reqser import request_to_dict, request_from_dict
from . import picklecompat
class Base(object):
"""Per-spider base queue class"""
def __init__(self, server, spider, key, serializer=None):
"""Initialize per-spider redis queue.
Parameters
----------
server : StrictRedis
Redis client instance.
spider : Spider
Scrapy spider instance.
key: str
Redis key where to put and get messages.
serializer : object
Serializer object with ``loads`` and ``dumps`` methods.
"""
if serializer is None:
# Backward compatibility.
# TODO: deprecate pickle.
serializer = picklecompat
# 当序列化 没有loads函数时 抛出异常
# 抛出异常的目的就是为了使传过来的序列化必须含有loads函数
if not hasattr(serializer, 'loads'):
raise TypeError("serializer does not implement 'loads' function: %r"
% serializer)
if not hasattr(serializer, 'dumps'):
raise TypeError("serializer '%s' does not implement 'dumps' function: %r"
% serializer)
# 下面的函数当类的所有函数 都可以使用
self.server = server
self.spider = spider
self.key = key % {'spider': spider.name}
self.serializer = serializer
# 将request进行编码成字符串
def _encode_request(self, request):
"""Encode a request object"""
# 将requests转换为字典
obj = request_to_dict(request, self.spider)
# 将字典转换为字符串进行返回
return self.serializer.dumps(obj)
# 将已经编码的encoded_request解码为字典
def _decode_request(self, encoded_request):
"""Decode an request previously encoded"""
obj = self.serializer.loads(encoded_request)
# 将dict转换为request objects 可以直接通过下载器进行下载
return request_from_dict(obj, self.spider)
# len方法 必须被重载 否则不能使用
def __len__(self):
"""Return the length of the queue"""
raise NotImplementedError
def push(self, request):
"""Push a request"""
raise NotImplementedError
def pop(self, timeout=0):
"""Pop a request"""
raise NotImplementedError
# 删除指定的self.key
def clear(self):
"""Clear queue/stack"""
self.server.delete(self.key)
首先看一下_encode_request() 和_decode_request()方法。我们把一个reques对象存储到数据库中，但数据库无法直接存储对象，多以要将request序列化转化为字符串，而这两个方法分别可以实现序列化和反序列化的操作，这个过程中可以利用pickle库来实现，队列queu在调用push()方法将request 存入数据库时，会调用_encode_request ()方法进行序列化，在调用pop()取出request时，会调用_decode_request()进行反序列化
