From：https://blog.csdn.net/weixin_37947156/article/details/75044971
From：https://cuiqingcai.com/6058.html

Scrapy-redis github：https://github.com/rmax/scrapy-redis
scrapy-redis分布式爬虫框架详解：https://segmentfault.com/a/1190000014333162?utm_source=channel-hottest
集群版 Scrapy-Redis：https://github.com/thsheep/scrapy_redis_cluster
scrapy-redis 和 scrapy 有什么区别？：https://www.zhihu.com/question/32302268
scrapy-redis使用以及剖析：https://www.cnblogs.com/wangyongsong/p/7485852.html
scrapy-redis 解析：https://www.cnblogs.com/zy0517/articles/9109681.html
基于 Scrapy-redis 的分布式爬虫设计：https://www.jianshu.com/p/cd4054bbc757/
小白进阶之Scrapy第三篇（基于Scrapy-Redis的分布式以及cookies池）：https://cuiqingcai.com/4048.html
Scrapy+redis实现分布式爬虫简易教程：https://www.jianshu.com/p/ed5afa658ccb?from=jiantop.com

scrapy 是 python 的一个非常好用的爬虫库，功能非常强大，如果是小站的话，我们使用 scrapy 本身就可以满足。但是当我们要爬取的页面非常多的时候，面对一些比较大型的站点的时候，单个 scrapy 就显得力不从心了。单个主机的处理能力就不能满足我们的需求了（无论是处理速度还是网络请求的并发数）。

这时候分布式爬虫的优势就显现出来，人多力量大。很遗憾 Scrapy 官方并不支持多个同时采集一个站点，虽然官方给出一个方法：**将一个站点的分割成几部分交给不同的scrapy去采集**。似乎是个解决办法，但是很麻烦诶！毕竟分割很麻烦的哇

下面就该 Scrapy-Redis 登场了。scrapy-redis 就是结合了分布式数据库 redis，重写了 scrapy 一些比较关键的代码，将 scrapy 变成一个可以在多个主机上同时运行的分布式爬虫。

scrapy-redis 是 github 上的一个开源项目，可以直接下载到他的源代码： https://github.com/rmax/scrapy-redis

scrapy-redis 的官方文档写的比较简洁，没有提及其运行原理，所以如果想全面的理解分布式爬虫的运行原理，还是得看 scrapy的源代码才行（还得先理解 scrapy 的运行原理，不然看 scrapy-redis 还是比较费劲）。

来看一看 Scrapy 的架构图

这张图大家相信大家都很熟悉了。重点看一下SCHEDULER

1. 先来看看官方对于SCHEDULER的定义：

**SCHEDULER接受来自Engine的Requests,并将它们放入队列（可以按顺序优先级），以便在之后将其提供给Engine**

官方文档：https://doc.scrapy.org/en/latest/topics/architecture.html#component-scheduler

2. 现在我们来看看SCHEDULER都提供了些什么功能：

根据官方文档说明在我们没有没有指定 SCHEDULER 参数时，默认使用：'scrapy.core.scheduler.Scheduler' 作为SCHEDULER(调度器)

scrapy.core.scheduler.py:

class Scheduler(object):def __init__(self, dupefilter, jobdir=None, dqclass=None, mqclass=None,logunser=False, stats=None, pqclass=None):self.df = dupefilterself.dqdir = self._dqdir(jobdir)self.pqclass = pqclassself.dqclass = dqclassself.mqclass = mqclassself.logunser = logunserself.stats = stats@classmethoddef from_crawler(cls, crawler):'''注意在 scrapy 中优先注意这个方法，此方法是一个钩子 用于访问当前爬虫的配置'''settings = crawler.settings# 获取去重用的类 默认：scrapy.dupefilters.RFPDupeFilterdupefilter_cls = load_object(settings['DUPEFILTER_CLASS'])# 对去重类进行配置from_settings 在 scrapy.dupefilters.RFPDupeFilter 43行# 这种调用方式对于IDE跳转不是很好  所以需要自己去找# @classmethod# def from_settings(cls, settings):#     debug = settings.getbool('DUPEFILTER_DEBUG')#     return cls(job_dir(settings), debug)# 上面就是from_settings方法 其实就是设置工作目录 和是否开启debugdupefilter = dupefilter_cls.from_settings(settings)# 获取优先级队列 类对象 默认：queuelib.pqueue.PriorityQueuepqclass = load_object(settings['SCHEDULER_PRIORITY_QUEUE'])# 获取磁盘队列 类对象（SCHEDULER使用磁盘存储 重启不会丢失）dqclass = load_object(settings['SCHEDULER_DISK_QUEUE'])# 获取内存队列 类对象（SCHEDULER使用内存存储 重启会丢失）mqclass = load_object(settings['SCHEDULER_MEMORY_QUEUE'])# 是否开启debuglogunser = settings.getbool('LOG_UNSERIALIZABLE_REQUESTS', settings.getbool('SCHEDULER_DEBUG'))# 将这些参数传递给 __init__方法return cls(dupefilter, jobdir=job_dir(settings), logunser=logunser,stats=crawler.stats, pqclass=pqclass, dqclass=dqclass, mqclass=mqclass)def has_pending_requests(self):"""检查是否有没处理的请求"""return len(self) > 0def open(self, spider):"""Engine创建完毕之后会调用这个方法"""self.spider = spider# 创建一个有优先级的内存队列 实例化对象# self.pqclass 默认是：queuelib.pqueue.PriorityQueue# self._newmq 会返回一个内存队列的 实例化对象 在110  111 行self.mqs = self.pqclass(self._newmq)# 如果self.dqdir 有设置 就创建一个磁盘队列 否则self.dqs 为空self.dqs = self._dq() if self.dqdir else None# 获得一个去重实例对象 open 方法是从BaseDupeFilter继承的# 现在我们可以用self.df来去重啦return self.df.open()def close(self, reason):"""当然Engine关闭时"""# 如果有磁盘队列 则对其进行dump后保存到active.json文件中if self.dqs:prios = self.dqs.close()with open(join(self.dqdir, 'active.json'), 'w') as f:json.dump(prios, f)# 然后关闭去重return self.df.close(reason)def enqueue_request(self, request):"""添加一个Requests进调度队列"""# self.df.request_seen是检查这个Request是否已经请求过了 如果有会返回Trueif not request.dont_filter and self.df.request_seen(request):# 如果Request的dont_filter属性没有设置（默认为False）和 已经存在则去重# 不push进队列self.df.log(request, self.spider)return False# 先尝试将Request push进磁盘队列dqok = self._dqpush(request)if dqok:# 如果成功 则在记录一次状态self.stats.inc_value('scheduler/enqueued/disk', spider=self.spider)else:# 不能添加进磁盘队列则会添加进内存队列self._mqpush(request)self.stats.inc_value('scheduler/enqueued/memory', spider=self.spider)self.stats.inc_value('scheduler/enqueued', spider=self.spider)return Truedef next_request(self):"""从队列中获取一个Request"""# 优先从内存队列中获取request = self.mqs.pop()if request:self.stats.inc_value('scheduler/dequeued/memory', spider=self.spider)else:# 不能获取的时候从磁盘队列队里获取request = self._dqpop()if request:self.stats.inc_value('scheduler/dequeued/disk', spider=self.spider)if request:self.stats.inc_value('scheduler/dequeued', spider=self.spider)# 将获取的到Request返回给Enginereturn requestdef __len__(self):return len(self.dqs) + len(self.mqs) if self.dqs else len(self.mqs)def _dqpush(self, request):if self.dqs is None:returntry:reqd = request_to_dict(request, self.spider)self.dqs.push(reqd, -request.priority)except ValueError as e:  # non serializable requestif self.logunser:msg = ("Unable to serialize request: %(request)s - reason:"" %(reason)s - no more unserializable requests will be"" logged (stats being collected)")logger.warning(msg, {'request': request, 'reason': e},exc_info=True, extra={'spider': self.spider})self.logunser = Falseself.stats.inc_value('scheduler/unserializable',spider=self.spider)returnelse:return Truedef _mqpush(self, request):self.mqs.push(request, -request.priority)def _dqpop(self):if self.dqs:d = self.dqs.pop()if d:return request_from_dict(d, self.spider)def _newmq(self, priority):return self.mqclass()def _newdq(self, priority):return self.dqclass(join(self.dqdir, 'p%s' % priority))def _dq(self):activef = join(self.dqdir, 'active.json')if exists(activef):with open(activef) as f:prios = json.load(f)else:prios = ()q = self.pqclass(self._newdq, startprios=prios)if q:logger.info("Resuming crawl (%(queuesize)d requests scheduled)",{'queuesize': len(q)}, extra={'spider': self.spider})return qdef _dqdir(self, jobdir):if jobdir:dqdir = join(jobdir, 'requests.queue')if not exists(dqdir):os.makedirs(dqdir)return dqdir

从上面的代码可以很清楚的知道 SCHEDULER 主要是完成了 push Request、pop Request 和去重的操作。而且 queue 操作是在内存队列中完成的。大家看 queuelib.queue 就会发现是基于内存的（deque）。

那么去重呢？

class RFPDupeFilter(BaseDupeFilter):"""Request Fingerprint duplicates filter"""def __init__(self, path=None, debug=False):self.file = Noneself.fingerprints = set()self.logdupes = Trueself.debug = debugself.logger = logging.getLogger(__name__)if path:# 此处可以看到去重其实打开了一个名叫 requests.seen的文件# 如果是使用的磁盘的话self.file = open(os.path.join(path, 'requests.seen'), 'a+')self.file.seek(0)self.fingerprints.update(x.rstrip() for x in self.file)@classmethoddef from_settings(cls, settings):debug = settings.getbool('DUPEFILTER_DEBUG')return cls(job_dir(settings), debug)def request_seen(self, request):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的时候，它们相互之间并不知道对方采集了些什么那些没有没采集。那就大家伙儿自己玩自己的了。完全没没有效率的提升啊！

怎么解决呢？

这就是我们 Scrapy-Redis 解决的问题了，不能协作不就是因为 Request 和去重这两个不能共享吗？

那我把这两个独立出来好了。

将 Scrapy 中的 SCHEDULER 组件独立放到大家都能访问的地方不就OK啦！加上 scrapy-redis 后流程图就应该变成这样了?

scrapy-redis 在 scrapy 的架构上增加了 redis，基于 redis 的特性拓展了如下四种组件：Scheduler，Duplication Filter，Item Pipeline，Base Spider

scrapy-redis 源码分析

scrapy-redis 的源代码很少，也比较好懂，很快就能看完。

下面开始 scrapy-redis 源码分析：

scrapy-redis 工程的主体还是 redis 和 scrapy 两个库，工程本身实现的东西不是很多，这个工程就像胶水一样，把这两个插件粘结了起来。下面我们来看看，scrapy-redis的每一个源代码文件都实现了什么功能，最后如何实现分布式的爬虫系统：

defaults.py

redis 的一些基础的默认的设置。其实就是一些默认配置：

import redis# For standalone use.
DUPEFILTER_KEY = 'dupefilter:%(timestamp)s'PIPELINE_KEY = '%(spider)s:items'REDIS_CLS = redis.StrictRedis
REDIS_ENCODING = 'utf-8'
# Sane connection defaults.
REDIS_PARAMS = {'socket_timeout': 30,'socket_connect_timeout': 30,'retry_on_timeout': True,'encoding': REDIS_ENCODING,
}SCHEDULER_QUEUE_KEY = '%(spider)s:requests'
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.PriorityQueue'
SCHEDULER_DUPEFILTER_KEY = '%(spider)s:dupefilter'
SCHEDULER_DUPEFILTER_CLASS = 'scrapy_redis.dupefilter.RFPDupeFilter'START_URLS_KEY = '%(name)s:start_urls'
START_URLS_AS_SET = False

connect.py

connect 文件引入了redis 模块，这个是 redis-python库的接口，用于通过python访问redis数据库，可见，这个文件主要是实现连接redis数据库的功能（返回的是redis库的Redis对象或者StrictRedis对象，这俩都是可以直接用来进行数据操作的对象）。这些连接接口在其他文件中经常被用到。其中，我们可以看到，要想连接到redis数据库，和其他数据库差不多，需要一个ip地址、端口号、用户名密码（可选）和一个整形的数据库编号，同时我们还可以在scrapy工程的setting文件中配置套接字的超时时间、等待时间等。

其实这个模块的功能：

1. 从 settings 里面获取 redis 的链接配置
2. 获取 redis 的链接实例

import sixfrom scrapy.utils.misc import load_objectfrom . import defaults# Shortcut maps 'setting name' -> 'parmater name'.
SETTINGS_PARAMS_MAP = {'REDIS_URL': 'url','REDIS_HOST': 'host','REDIS_PORT': 'port','REDIS_ENCODING': 'encoding',
}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. Youcan override them using the ``REDIS_PARAMS`` setting.Parameters----------settings : SettingsA scrapy settings object. See the supported settings below.Returns-------serverRedis client instance.Other Parameters----------------REDIS_URL : str, optionalServer connection URL.REDIS_HOST : str, optionalServer host.REDIS_PORT : str, optionalServer port.REDIS_ENCODING : str, optionalData encoding.REDIS_PARAMS : dict, optionalAdditional client parameters."""params = defaults.REDIS_PARAMS.copy()params.update(settings.getdict('REDIS_PARAMS'))# XXX: Deprecate REDIS_* settings.for source, dest in SETTINGS_PARAMS_MAP.items():val = settings.get(source)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_settingsdef get_redis(**kwargs):"""Returns a redis client instance.Parameters----------redis_cls : class, optionalDefaults to ``redis.StrictRedis``.url : str, optionalIf given, ``redis_cls.from_url`` is used to instantiate the class.**kwargsExtra parameters to be passed to the ``redis_cls`` class.Returns-------serverRedis client instance."""redis_cls = kwargs.pop('redis_cls', defaults.REDIS_CLS)url = kwargs.pop('url', None)if url:return redis_cls.from_url(url, **kwargs)else:return redis_cls(**kwargs)

dupefilters.py

这个主要是用来去重的。RFPDupeFilter继承自 Scrapy 的BaseDupeFilter，实现了 request 去重功能，基于 Scrapy 的 request_fingerprint 生成指纹，并在 Redis 上存储。当收到新的 request，首先生成指纹判断是否存在于已爬取的指纹库内（Redis set），若存在则返回 False，不存在返回 True.总得来说是这样的，这个文件首先获取到redis的server,然后从scrapy的request中获取request的指纹，将这个指纹进行存到redis的去重库中。达到去重的目的。

这个文件看起来比较复杂，重写了scrapy本身已经实现的 request 判重功能。因为本身 scrapy 单机跑的话，只需要读取内存中的request 队列或者持久化的 request 队列（scrapy默认的持久化似乎是json格式的文件，不是数据库）就能判断这次要发出的request url是否已经请求过或者正在调度（本地读就行了）。而 分布式跑的话，就需要各个主机上的scheduler都连接同一个数据库的同一个 request池来判断这次的请求是否是重复的了。

在这个文件中，通过继承 BaseDupeFilter 重写他的方法，实现了基于redis的判重。根据源代码来看，scrapy-redis 使用了scrapy本身的一个 fingerprint 接口 request_fingerprint，这个接口很有趣，根据scrapy文档所说，他通过hash来判断两个url是否相同（相同的url会生成相同的hash结果），但是当两个url的地址相同，get型参数相同但是顺序不同时，也会生成相同的hash结果（这个真的比较神奇。。。）所以 scrapy-redis 依旧使用 url 的 fingerprint 来判断 request 请求是否已经出现过。这个类通过连接 redis，使用一个key来向redis的一个set中插入fingerprint（这个key对于同一种spider是相同的，redis 是一个key-value的数据库，如果key是相同的，访问到的值就是相同的，这里使用 spider名字+DupeFilter 的 key 就是为了在不同主机上的不同爬虫实例，只要属于同一种 spider，就会访问到同一个set，而这个 set 就是他们的url判重池），如果返回值为0，说明该set中该fingerprint 已经存在（因为集合是没有重复值的），则返回 False，如果返回值为 1，说明添加了一个fingerprint到set中，则说明这个 request 没有重复，于是返回True，还顺便把新fingerprint加入到数据库中了。

DupeFilter 判重会在 scheduler 类中用到，每一个 request 在进入调度之前都要进行判重，如果重复就不需要参加调度，直接舍弃就好了，不然就是白白浪费资源。

import logging
import timefrom scrapy.dupefilters import BaseDupeFilter
from scrapy.utils.request import request_fingerprintfrom . import defaults
from .connection import get_redis_from_settingslogger = 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 = loggerdef __init__(self, server, key, debug=False):"""Initialize the duplicates filter.Parameters----------server : redis.StrictRedisThe redis server instance.key : strRedis key Where to store fingerprints.debug : bool, optionalWhether to log filtered requests."""self.server = serverself.key = keyself.debug = debugself.logdupes = True@classmethoddef 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 asit needs to pass the spider name in the key.Parameters----------settings : scrapy.settings.SettingsReturns-------RFPDupeFilterA RFPDupeFilter instance."""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 = defaults.DUPEFILTER_KEY % {'timestamp': int(time.time())}debug = settings.getbool('DUPEFILTER_DEBUG')return cls(server, key=key, debug=debug)@classmethoddef from_crawler(cls, crawler):"""Returns instance from crawler.Parameters----------crawler : scrapy.crawler.CrawlerReturns-------RFPDupeFilterInstance of RFPDupeFilter."""return cls.from_settings(crawler.settings)def request_seen(self, request):"""Returns True if request was already seen.Parameters----------request : scrapy.http.RequestReturns-------bool"""fp = self.request_fingerprint(request)# This returns the number of values added, zero if already exists.added = self.server.sadd(self.key, fp)return added == 0def request_fingerprint(self, request):"""Returns a fingerprint for a given request.Parameters----------request : scrapy.http.RequestReturns-------str"""return request_fingerprint(request)@classmethoddef from_spider(cls, spider):settings = spider.settingsserver = 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.Requestspider : 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

Scrapy 用集合实现这个 request 去重功能，Scrapy 中把已经发送的 request 指纹放入到一个集合中，把下一个request 的指纹拿到集合中比对，如果该指纹存在于集合中，说明这个 request 发送过了，如果没有则继续操作。

核心的判重功能：

在 scrapy-redis 中去重是由 Duplication Filter 组件来实现的，它通过 redis 的 set 不重复的特性，巧妙的实现了 DuplicationFilter去重。scrapy-redis 调度器从引擎接受request，将 request 的指纹存入 redis 的 set 检查是否重复，并将不重复的 request push写入 redis 的 request queue。

引擎请求 request (Spider发出的）时，调度器从 redis 的request queue 队列里根据优先级 pop 出⼀个request 返回给引擎，引擎将此request发给spider处理。

picklecompat.py

这里实现了 loads 和 dumps 两个函数，其实就是实现了一个 serializer，因为 redis 数据库不能存储复杂对象（value部分只能是字符串，字符串列表，字符串集合和hash，key部分只能是字符串），所以我们存啥都要先串行化成文本才行。这里使用的就是python 的 pickle 模块，一个兼容 py2 和 py3 的串行化工具。这个 serializer 主要用于一会的 scheduler 存 reuqest 对象，至于为什么不实用 json 格式，我也不是很懂，item pipeline 的串行化默认用的就是 json。

"""A pickle wrapper module with protocol=-1 by default."""try:import cPickle as pickle  # PY2
except ImportError:import pickledef loads(s):return pickle.loads(s)def dumps(obj):return pickle.dumps(obj, protocol=-1)

pipeline.py

这是是用来实现分布式处理的作用。它将 Item 存储在 redis 中以实现分布式处理。由于在这里需要读取配置，所以就用到了from_crawler() 函数。pipeline 文件实现了一个 item pipieline类，和 scrapy 的 item pipeline 是同一个对象，通过从 settings 中拿到我们配置的REDIS_ITEMS_KEY 作为 key，把 item 串行化之后存入 redis 数据库对应的 value 中（这个value可以看出出是个list，我们的每个item是这个list中的一个结点），这个pipeline把提取出的item存起来，主要是为了方便我们延后处理数据。

from scrapy.utils.misc import load_object
from scrapy.utils.serialize import ScrapyJSONEncoder
from twisted.internet.threads import deferToThreadfrom . import connection, defaultsdefault_serialize = ScrapyJSONEncoder().encodeclass RedisPipeline(object):"""Pushes serialized item into a redis list/queueSettings--------REDIS_ITEMS_KEY : strRedis key where to store items.REDIS_ITEMS_SERIALIZER : strObject path to serializer function."""def __init__(self, server,key=defaults.PIPELINE_KEY,serialize_func=default_serialize):"""Initialize pipeline.Parameters----------server : StrictRedisRedis client instance.key : strRedis key where to store items.serialize_func : callableItems serializer function."""self.server = serverself.key = keyself.serialize = serialize_func@classmethoddef from_settings(cls, settings):params = {'server': connection.from_settings(settings),}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)@classmethoddef from_crawler(cls, crawler):return cls.from_settings(crawler.settings)def process_item(self, item, spider):return deferToThread(self._process_item, item, spider)def _process_item(self, item, spider):key = self.item_key(item, spider)data = self.serialize(item)self.server.rpush(key, data)return itemdef item_key(self, item, spider):"""Returns redis key based on given spider.Override this function to use a different key depending on the itemand/or spider."""return self.key % {'spider': spider.name}

queue.py

该文件实现了几个容器类，可以看这些容器和redis交互频繁，同时使用了我们上边 picklecompat 中定义的 serializer。这个文件实现的几个容器大体相同，只不过一个是队列，一个是栈，一个是优先级队列，这三个容器到时候会被scheduler对象实例化，来实现 request的调度。比如：我们使用 SpiderQueue 作为调度队列的类型，到时候 request 的调度方法就是先进先出，而实用SpiderStack 就是先进后出了。

我们可以仔细看看 SpiderQueue 的实现，他的 push 函数就和其他容器的一样，只不过 push进去的 request请求先被scrapy的接口 request_to_dict 变成了一个dict对象（因为request对象实在是比较复杂，有方法有属性不好串行化），之后使用picklecompat中的serializer串行化为字符串，然后使用一个特定的 key 存入redis中（该key在同一种spider中是相同的）。而调用pop时，其实就是从redis用那个特定的key去读其值（一个list），从list中读取最早进去的那个，于是就先进先出了。

这些容器类都会作为 scheduler 调度 request 的容器，scheduler 在每个主机上都会实例化一个，并且和 spider一一对应，所以分布式运行时会有一个 spider 的多个实例和一个 scheduler 的多个实例存在于不同的主机上，但是，因为 scheduler 都是用相同的容器，而这些容器都连接同一个 redis 服务器，又都使用 spider 名加 queue 来作为 key 读写数据，所以不同主机上的不同爬虫实例公用一个 request 调度池，实现了分布式爬虫之间的统一调度。

from scrapy.utils.reqser import request_to_dict, request_from_dictfrom . import picklecompatclass Base(object):"""Per-spider base queue class"""def __init__(self, server, spider, key, serializer=None):"""Initialize per-spider redis queue.Parameters----------server : StrictRedisRedis client instance.spider : SpiderScrapy spider instance.key: strRedis key where to put and get messages.serializer : objectSerializer object with ``loads`` and ``dumps`` methods."""if serializer is None:# Backward compatibility.# TODO: deprecate pickle.serializer = picklecompatif 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 = serverself.spider = spiderself.key = key % {'spider': spider.name}self.serializer = serializerdef _encode_request(self, request):"""Encode a request object"""obj = request_to_dict(request, self.spider)return self.serializer.dumps(obj)def _decode_request(self, encoded_request):"""Decode an request previously encoded"""obj = self.serializer.loads(encoded_request)return request_from_dict(obj, self.spider)def __len__(self):"""Return the length of the queue"""raise NotImplementedErrordef push(self, request):"""Push a request"""raise NotImplementedErrordef pop(self, timeout=0):"""Pop a request"""raise NotImplementedErrordef clear(self):"""Clear queue/stack"""self.server.delete(self.key)class FifoQueue(Base):"""Per-spider FIFO queue"""def __len__(self):"""Return the length of the queue"""return self.server.llen(self.key)def push(self, request):"""Push a request"""self.server.lpush(self.key, self._encode_request(request))def pop(self, timeout=0):"""Pop a request"""if timeout > 0:data = self.server.brpop(self.key, timeout)if isinstance(data, tuple):data = data[1]else:data = self.server.rpop(self.key)if data:return self._decode_request(data)class PriorityQueue(Base):"""Per-spider priority queue abstraction using redis' sorted set"""def __len__(self):"""Return the length of the queue"""return self.server.zcard(self.key)def push(self, request):"""Push a request"""data = self._encode_request(request)score = -request.priority# We don't use zadd method as the order of arguments change depending on# whether the class is Redis or StrictRedis, and the option of using# kwargs only accepts strings, not bytes.self.server.execute_command('ZADD', self.key, score, data)def pop(self, timeout=0):"""Pop a requesttimeout not support in this queue class"""# use atomic range/remove using multi/execpipe = self.server.pipeline()pipe.multi()pipe.zrange(self.key, 0, 0).zremrangebyrank(self.key, 0, 0)results, count = pipe.execute()if results:return self._decode_request(results[0])class LifoQueue(Base):"""Per-spider LIFO queue."""def __len__(self):"""Return the length of the stack"""return self.server.llen(self.key)def push(self, request):"""Push a request"""self.server.lpush(self.key, self._encode_request(request))def pop(self, timeout=0):"""Pop a request"""if timeout > 0:data = self.server.blpop(self.key, timeout)if isinstance(data, tuple):data = data[1]else:data = self.server.lpop(self.key)if data:return self._decode_request(data)# TODO: Deprecate the use of these names.
SpiderQueue = FifoQueue
SpiderStack = LifoQueue
SpiderPriorityQueue = PriorityQueue

可以看出，是以 base 为基类，然后被三个队列的类继承。然后进行了pop和push的操作。

FifoQueue: 继承 Base，重写了 push 和 pop 方法实现 先进先出 队列。
PriorityQueue: 继承 Base，重写了 push 和 pop 方法实现 优先级 队列。
LifoQueue: 继承 Base，重写了 push 和 pop 方法实现 后进先出 队列。

scheduler.py

scrapy 改造了 python 本来的 collection.deque(双向队列)形成了自己的 Scrapy queue，但是 Scrapy 多个 spider 不能共享待爬取队列 Scrapy queue，即 Scrapy 本身不支持爬虫分布式。

scrapy-redis 的解决是把这个 Scrapy queue 换成 redis 数据库（也是指 redis 队列），从同一个 redis-server 存放要爬取的request，便能让多个 spider 去同一个数据库里读取。

Scrapy 中跟 “待爬队列” 直接相关的就是调度器 Scheduler，它负责对新的 request 进行入列操作（加入Scrapy queue），取出下一个要爬取的 request（从Scrapy queue中取出）等操作。它把待爬队列按照优先级建立了一个字典结构，然后根据 request 中的优先级，来决定该入哪个队列，出列时则按优先级较小的优先出列。为了管理这个比较高级的队列字典，Scheduler 需要提供一系列的方法。但是原来的 Scheduler 已经无法使用，所以使用 Scrapy-redis 的 scheduler 组件。

scheduler.py 这个文件重写了 scheduler 类，用来代替 scrapy.core.scheduler 的原有调度器。其实对原有调度器的逻辑没有很大的改变，主要是使用了redis 作为数据存储的媒介，以达到各个爬虫之间的统一调度。 scheduler 负责调度各个 spider 的 request 请求，scheduler 初始化时，通过 settings 文件读取 queue 和 dupefilters 的类型（一般就用上边默认的），配置 queue 和 dupefilters 使用的 key（一般就是spider name加上queue或者dupefilters，这样对于同一种spider 的不同实例，就会使用相同的数据块了）。每当一个 request 要被调度时，enqueue_request 被调用，scheduler 使用dupefilters 来判断这个url是否重复，如果不重复，就添加到 queue 的容器中（先进先出，先进后出和优先级都可以，可以在settings中配置）。当调度完成时，next_request 被调用，scheduler 就通过 queue 容器的接口，取出一个 request，把他发送给相应的 spider，让spider 进行爬取工作。

import importlib
import sixfrom scrapy.utils.misc import load_objectfrom . import connection, defaults# TODO: add SCRAPY_JOB support.
class Scheduler(object):"""Redis-based schedulerSettings--------SCHEDULER_PERSIST : bool (default: False)Whether to persist or clear redis queue.SCHEDULER_FLUSH_ON_START : bool (default: False)Whether to flush redis queue on start.SCHEDULER_IDLE_BEFORE_CLOSE : int (default: 0)How many seconds to wait before closing if no message is received.SCHEDULER_QUEUE_KEY : strScheduler redis key.SCHEDULER_QUEUE_CLASS : strScheduler queue class.SCHEDULER_DUPEFILTER_KEY : strScheduler dupefilter redis key.SCHEDULER_DUPEFILTER_CLASS : strScheduler dupefilter class.SCHEDULER_SERIALIZER : strScheduler serializer."""def __init__(self, server,persist=False,flush_on_start=False,queue_key=defaults.SCHEDULER_QUEUE_KEY,queue_cls=defaults.SCHEDULER_QUEUE_CLASS,dupefilter_key=defaults.SCHEDULER_DUPEFILTER_KEY,dupefilter_cls=defaults.SCHEDULER_DUPEFILTER_CLASS,idle_before_close=0,serializer=None):"""Initialize scheduler.Parameters----------server : RedisThe redis server instance.persist : boolWhether to flush requests when closing. Default is False.flush_on_start : boolWhether to flush requests on start. Default is False.queue_key : strRequests queue key.queue_cls : strImportable path to the queue class.dupefilter_key : strDuplicates filter key.dupefilter_cls : strImportable path to the dupefilter class.idle_before_close : intTimeout before giving up."""if idle_before_close < 0:raise TypeError("idle_before_close cannot be negative")self.server = serverself.persist = persistself.flush_on_start = flush_on_startself.queue_key = queue_keyself.queue_cls = queue_clsself.dupefilter_cls = dupefilter_clsself.dupefilter_key = dupefilter_keyself.idle_before_close = idle_before_closeself.serializer = serializerself.stats = Nonedef __len__(self):return len(self.queue)@classmethoddef from_settings(cls, settings):kwargs = {'persist': settings.getbool('SCHEDULER_PERSIST'),'flush_on_start': settings.getbool('SCHEDULER_FLUSH_ON_START'),'idle_before_close': settings.getint('SCHEDULER_IDLE_BEFORE_CLOSE'),}# If these values are missing, it means we want to use the defaults.optional = {# TODO: Use custom prefixes for this settings to note that are# specific to scrapy-redis.'queue_key': 'SCHEDULER_QUEUE_KEY','queue_cls': 'SCHEDULER_QUEUE_CLASS','dupefilter_key': 'SCHEDULER_DUPEFILTER_KEY',# We use the default setting name to keep compatibility.'dupefilter_cls': 'DUPEFILTER_CLASS','serializer': 'SCHEDULER_SERIALIZER',}for name, setting_name in optional.items():val = settings.get(setting_name)if val:kwargs[name] = val# Support serializer as a path to a module.if isinstance(kwargs.get('serializer'), six.string_types):kwargs['serializer'] = importlib.import_module(kwargs['serializer'])server = connection.from_settings(settings)# Ensure the connection is working.server.ping()return cls(server=server, **kwargs)@classmethoddef from_crawler(cls, crawler):instance = cls.from_settings(crawler.settings)# FIXME: for now, stats are only supported from this constructorinstance.stats = crawler.statsreturn instancedef open(self, spider):self.spider = spidertry:self.queue = load_object(self.queue_cls)(server=self.server,spider=spider,key=self.queue_key % {'spider': spider.name},serializer=self.serializer,)except TypeError as e:raise ValueError("Failed to instantiate queue class '%s': %s",self.queue_cls, e)self.df = load_object(self.dupefilter_cls).from_spider(spider)if self.flush_on_start:self.flush()# notice if there are requests already in the queue to resume the crawlif len(self.queue):spider.log("Resuming crawl (%d requests scheduled)" % len(self.queue))def close(self, reason):if not self.persist:self.flush()def flush(self):self.df.clear()self.queue.clear()def enqueue_request(self, request):if not request.dont_filter and self.df.request_seen(request):self.df.log(request, self.spider)return Falseif self.stats:self.stats.inc_value('scheduler/enqueued/redis', spider=self.spider)self.queue.push(request)return Truedef next_request(self):block_pop_timeout = self.idle_before_closerequest = self.queue.pop(block_pop_timeout)if request and self.stats:self.stats.inc_value('scheduler/dequeued/redis', spider=self.spider)return requestdef has_pending_requests(self):return len(self) > 0

这个文件下只有一个类 Scheduler，一如既往的通过类方法来实例化来实现外部可以直接通过调用两个方法然后从爬虫中获取settings 和 crawler，但是有两个比较特殊的函数是 def open(self, spider) 和 def next_request(self)：

open(): 调度器启动时的自动的操作，这里主要实例化了任务队列 queue 和过滤器 dupefilter。
next_request(): 从任务队列取出 request。

spider.py

如果在 settings.py 里面：REDIS_START_URLS_AS_SET = False 的话，就是列表的形式，存入就是 lpush 或者是 rpush 等操作，如果是 REDIS_START_URLS_AS_SET = True 的话，那么存入就是集合的形式，sadd 等操作。

REDIS_START_URLS_AS_SET = False # 默认是 False ，即默认从列表的格式取数据出来。

如果不设置 REDIS_START_URLS_KEY，则默认 REDIS_START_URLS_KEY = '%(name)s:start_urls' ，这个是存入 redis 里面的 key，可以根据这来取 value，例如：start_urls:baidu

redis 写入 URL （即添加任务）：

添加列表形式的任务：

import  redisconn = redis.Redis(host='127.0.0.1',port=6379)
conn.lpush('start_urls:baidu','http://www.baidu.com')

添加集合形式的任务：

import redisconn = redis.Redis(host='127.0.0.1', port=6379)
conn.sadd('start_urls:baidu', 'http://www.baidu.com')  # 按照这个格式来存数据的
print(conn.smembers('start_urls:baidu'))

使用 scrapy-redis 提供的方法添加任务示例：

首先需要在 setting.py 里面配置：

# REDIS_URL = 'redis://用户名:密码@主机IP:端口'
REDIS_URL = 'redis://127.0.0.1:6379'
REDIS_PARAMS = dict(db=15)

添加任务：

import redis
from scrapy.utils.project import get_project_settings
from scrapy_redis.connection import get_redis_from_settings# 方法 1
# 如果是在 scrapy-redis 工程里面可以使用这个方法
server_1 = get_redis_from_settings(get_project_settings())
server_1.sadd('start_urls:server_1', 'http://www.baidu.com')# 方法 2
server_2 = redis.Redis(host='127.0.0.1', port=6379, db=15)
server_2.sadd('start_urls:server_2', 'http://www.baidu.com')# 方法 3
redis_url = 'redis://root:xxxx@47.110.xx.xx:6379'# 加上 decode_responses=True，写入的键值对中的value为str类型，不加这个参数写入的则为字节类型。
r = redis.Redis.from_url(redis_url, decode_responses=True) 
r.lpush('test_key', 'wwww')

spider 的改动也不是很大，主要是通过 connect 接口，给 spider 绑定了 spider_idle 信号，spider 初始化时，通过 setup_redis 函数初始化好 redis 的连接，之后通过 next_requests 函数从 redis 中取出 strat url，使用的 key 是 settings 中REDIS_START_URLS_AS_SET 定义的（注意了这里的初始化 url 池和我们上边的 queue 的 url池不是一个东西，queue的池是用于调度的，初始化 url池是存放入口 url 的，他们都存在 redis 中，但是使用不同的 key 来区分，就当成是不同的表吧），spider 使用少量的 start url，可以发展出很多新的 url，这些 url 会进入 scheduler 进行判重和调度。直到 spider 跑到调度池内没有 url 的时候，会触发 spider_idle 信号，从而触发 spider 的 next_requests 函数，再次从 redis 的 start url 池中读取一些url。

分析：在这个 spider 中通过 connect signals.spider_idle 信号实现对 crawler 状态的监视。当 idle 时，返回新的make_requests_from_url(url) 给引擎，进而交给调度器调度。

from scrapy import signals
from scrapy.exceptions import DontCloseSpider
from scrapy.spiders import Spider, CrawlSpiderfrom . import connection, defaults
from .utils import bytes_to_strclass RedisMixin(object):"""Mixin class to implement reading urls from a redis queue."""redis_key = Noneredis_batch_size = Noneredis_encoding = None# Redis client placeholder.server = Nonedef start_requests(self):"""Returns a batch of start requests from redis."""return self.next_requests()def setup_redis(self, crawler=None):"""Setup redis connection and idle signal.This should be called after the spider has set its crawler object."""if self.server is not None:returnif crawler is None:# We allow optional crawler argument to keep backwards# compatibility.# XXX: Raise a deprecation warning.crawler = getattr(self, 'crawler', None)if crawler is None:raise ValueError("crawler is required")settings = crawler.settingsif self.redis_key is None:self.redis_key = settings.get('REDIS_START_URLS_KEY', defaults.START_URLS_KEY,)self.redis_key = self.redis_key % {'name': self.name}if not self.redis_key.strip():raise ValueError("redis_key must not be empty")if self.redis_batch_size is None:# TODO: Deprecate this setting (REDIS_START_URLS_BATCH_SIZE).self.redis_batch_size = settings.getint('REDIS_START_URLS_BATCH_SIZE',settings.getint('CONCURRENT_REQUESTS'),)try:self.redis_batch_size = int(self.redis_batch_size)except (TypeError, ValueError):raise ValueError("redis_batch_size must be an integer")if self.redis_encoding is None:self.redis_encoding = settings.get('REDIS_ENCODING', defaults.REDIS_ENCODING)self.logger.info("Reading start URLs from redis key '%(redis_key)s' ""(batch size: %(redis_batch_size)s, encoding: %(redis_encoding)s",self.__dict__)self.server = connection.from_settings(crawler.settings)# The idle signal is called when the spider has no requests left,# that's when we will schedule new requests from redis queuecrawler.signals.connect(self.spider_idle, signal=signals.spider_idle)def next_requests(self):"""Returns a request to be scheduled or none."""use_set = self.settings.getbool('REDIS_START_URLS_AS_SET', defaults.START_URLS_AS_SET)fetch_one = self.server.spop if use_set else self.server.lpop# XXX: Do we need to use a timeout here?found = 0# TODO: Use redis pipeline execution.while found < self.redis_batch_size:data = fetch_one(self.redis_key)if not data:# Queue empty.breakreq = self.make_request_from_data(data)if req:yield reqfound += 1else:self.logger.debug("Request not made from data: %r", data)if found:self.logger.debug("Read %s requests from '%s'", found, self.redis_key)def make_request_from_data(self, data):"""Returns a Request instance from data coming from Redis.By default, ``data`` is an encoded URL. You can override this method toprovide your own message decoding.Parameters----------data : bytesMessage from redis."""url = bytes_to_str(data, self.redis_encoding)return self.make_requests_from_url(url)def schedule_next_requests(self):"""Schedules a request if available"""# TODO: While there is capacity, schedule a batch of redis requests.for req in self.next_requests():self.crawler.engine.crawl(req, spider=self)def spider_idle(self):"""Schedules a request if available, otherwise waits."""# XXX: Handle a sentinel to close the spider.self.schedule_next_requests()raise DontCloseSpiderclass RedisSpider(RedisMixin, Spider):"""Spider that reads urls from redis queue when idle.Attributes----------redis_key : str (default: REDIS_START_URLS_KEY)Redis key where to fetch start URLs from..redis_batch_size : int (default: CONCURRENT_REQUESTS)Number of messages to fetch from redis on each attempt.redis_encoding : str (default: REDIS_ENCODING)Encoding to use when decoding messages from redis queue.Settings--------REDIS_START_URLS_KEY : str (default: "<spider.name>:start_urls")Default Redis key where to fetch start URLs from..REDIS_START_URLS_BATCH_SIZE : int (deprecated by CONCURRENT_REQUESTS)Default number of messages to fetch from redis on each attempt.REDIS_START_URLS_AS_SET : bool (default: False)Use SET operations to retrieve messages from the redis queue. If False,the messages are retrieve using the LPOP command.REDIS_ENCODING : str (default: "utf-8")Default encoding to use when decoding messages from redis queue."""@classmethoddef from_crawler(self, crawler, *args, **kwargs):obj = super(RedisSpider, self).from_crawler(crawler, *args, **kwargs)obj.setup_redis(crawler)return objclass RedisCrawlSpider(RedisMixin, CrawlSpider):"""Spider that reads urls from redis queue when idle.Attributes----------redis_key : str (default: REDIS_START_URLS_KEY)Redis key where to fetch start URLs from..redis_batch_size : int (default: CONCURRENT_REQUESTS)Number of messages to fetch from redis on each attempt.redis_encoding : str (default: REDIS_ENCODING)Encoding to use when decoding messages from redis queue.Settings--------REDIS_START_URLS_KEY : str (default: "<spider.name>:start_urls")Default Redis key where to fetch start URLs from..REDIS_START_URLS_BATCH_SIZE : int (deprecated by CONCURRENT_REQUESTS)Default number of messages to fetch from redis on each attempt.REDIS_START_URLS_AS_SET : bool (default: True)Use SET operations to retrieve messages from the redis queue.REDIS_ENCODING : str (default: "utf-8")Default encoding to use when decoding messages from redis queue."""@classmethoddef from_crawler(self, crawler, *args, **kwargs):obj = super(RedisCrawlSpider, self).from_crawler(crawler, *args, **kwargs)obj.setup_redis(crawler)return obj

这个 spider 文件有三个类，RedisMixin 是一个基类，剩余两个是多继承。

RedisMixin 类：

1 setup_redis 主要是获取 redis 的 server 和获取爬虫的 idle signal，当爬虫没有请求时，调用空闲信号，那时我们将从 redis队列安排新请求。
2 spider_idle 主要是 idle 信号处理，这里调用 schedule_next_requests 完成从 Redis 调度
3 make_request_from_data 主要是从 redis 队列获取ur
4 next_requests 这个主要是 redis 获取 url，我从这里看到可以设置 url 队列在 redis 的存储的数据格式函数schedule_next_requests从next_requests 获取 url 包装为 HttpRequest

RedisSpider(RedisMixin, Spiser) 类：多继承，用 RedisMixin 调度功能覆盖 Spider 原生

RedisCrawlSpider(RedisMixin, CrawlSpiser) 类：多继承，用 RedisMixin 调度功能覆盖 CrawlSpider 原生

当编写分布式爬虫时，不在使用 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 读取 strart url 池，拿去新的一批 url 入口，然后再次重复上边的工作。

集成 bloomfilter 到 scrapy-redis 中

传送门：bloomfilter算法详解及实例

算法实现：bloomfilter_imooc

dupefilter.py：

import logging
import timefrom scrapy.dupefilters import BaseDupeFilter
from scrapy.utils.request import request_fingerprintfrom . import defaults
from .connection import get_redis_from_settingslogger = 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 = loggerdef __init__(self, server, key, debug=False):"""Initialize the duplicates filter.Parameters----------server : redis.StrictRedisThe redis server instance.key : strRedis key Where to store fingerprints.debug : bool, optionalWhether to log filtered requests."""self.server = serverself.key = keyself.debug = debugself.logdupes = True@classmethoddef 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 asit needs to pass the spider name in the key.Parameters----------settings : scrapy.settings.SettingsReturns-------RFPDupeFilterA RFPDupeFilter instance."""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 = defaults.DUPEFILTER_KEY % {'timestamp': int(time.time())}debug = settings.getbool('DUPEFILTER_DEBUG')return cls(server, key=key, debug=debug)@classmethoddef from_crawler(cls, crawler):"""Returns instance from crawler.Parameters----------crawler : scrapy.crawler.CrawlerReturns-------RFPDupeFilterInstance of RFPDupeFilter."""return cls.from_settings(crawler.settings)def request_seen(self, request):"""Returns True if request was already seen.Parameters----------request : scrapy.http.RequestReturns-------bool"""fp = self.request_fingerprint(request)# This returns the number of values added, zero if already exists.added = self.server.sadd(self.key, fp)return added == 0def request_fingerprint(self, request):"""Returns a fingerprint for a given request.Parameters----------request : scrapy.http.RequestReturns-------str"""return request_fingerprint(request)@classmethoddef from_spider(cls, spider):settings = spider.settingsserver = 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.Requestspider : 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

Scrapy-Redis 分布式爬虫框架使用实例

1. 创建项目

scrapy startproject example

example/
├── scrapy.cfg
└── example
├── __init__.py
├── items.py
├── middlewares.py
├── pipelines.py
├── settings.py
└── spiders
├── __init__.py
└── my_crawlspider.py/my_redisspider.py/my_rediscrawlspider.py # 3中类型的Scrap-Redis爬虫

2. 明确目标

vim items.py ：定义要爬取的字段

import scrapyclass ExampleItem(scrapy.Item):name = scrapy.Field()description = scrapy.Field()link = scrapy.Field()crawled = scrapy.Field()spider = scrapy.Field()url = scrapy.Field()

也可以不写 item ，直接返回一个 Python 类型的 dict 对象，因为 item 本身就是一个 dict 类型的对象。

3. 编写自定义 pipeline

vim pipelines.py

from datetime import datetime
class ExampPipeline(object):def process_item(self,item,spider):item['crawled'] = datetime.utcnow()  # 调用datetime.utcnow()方法获取爬虫执行时的UTC时间# 调用spider.name属性获取当前爬虫名(因为可能同时有多个爬虫在爬取,这样可以看到谁爬了哪些网页)item['spider'] = spider.name          return item

4. 注册自定义 pipeline 及 Scrapy-Redis 分布式爬虫相关设置

vim settings.py

#-----------Scrapy-Redis分布式爬虫相关设置如下-------------
# 使用Scrapy-Redis的去重组件,不再使用scrapy的去重组件
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"   
# 使用Scrapy-Redis的调度器,不再使用scrapy的调度器
SCHEDULER = "scrapy_redis.scheduler.Scheduler"               
# 使用Scrapy-Redis的从请求集合中取出请求的方式,三种方式择其一即可:
SCHEDULER_CLASS = "scrapy_redis.queue.SpiderPriorityQueue"   
# 分别按(1)请求的优先级/(2)队列FIFO/(3)栈FILO 取出请求
#SCHEDULER_CLASS = "scrapy_redis.queue.SpiderQueue"          
#SCHEDULER_CLASS = "scrapy_redis.queue.SpiderStack"
SCHEDULER_PERSIST = True         # 允许暂停,redis请求记录不会丢失(重启爬虫不会重头爬取已爬过的页面)REDIS_HOST = "200.200.200.200"   # 这两项是Redis连接设置,如果注释或不写会默认将数据存放到本机的Redis中
REDIS_PORT = 6379                # 注意:master端的Redis需要允许远程连接--配置中注释掉bind 127.0.0.1#----------注册RedisPipeline/自定义pipeline------------------
# # 注意:自定义pipeline的优先级需高于Redispipeline,因为RedisPipeline不会返回item,
# 所以如果RedisPipeline优先级高于自定义pipeline,那么自定义pipeline无法获取到item
ITEM_PIPELINES = {"example.pipelines.ExampPipeline":300,        # 自定义pipeline视情况选择性注册(可选)"scrapy_redis.pipelines.RedisPipeline":400    # 将RedisPipeline注册到pipeline组件中(这样才能将数据存入Redis)
}

5. 编写爬虫

( 三种 Scrapy-Redis 爬虫：CrawlSpider / RedisSpider / RedisCrawlSpider )

scrapy.Spider 和scrapy.CrawlSpider 区别：http://scrapy-chs.readthedocs.io/zh_CN/1.0/topics/spiders.html

Spider 是最简单的 spider。每个其他的 spider 必须继承自该类(包括 Scrapy 自带的其他 spider 以及您自己编写的 spider )。 Spider 并没有提供什么特殊的功能。其仅仅提供了 start_requests() 的默认实现，读取并请求 spider 属性中的 start_urls，并根据返回的结果 (resulting responses) 调用 spider 的 parse 方法。

CrawlSpider 是爬取一般网站常用的 spider。CrawlSpider 定义了一些规则(rule)来提供跟进 link 的方便的机制。也许该 spider并不是完全适合您的特定网站或项目，但其对很多情况都使用。因此您可以以其为起点，根据需求修改部分方法。当然您也可以实现自己的spider。

类型一：（基于 scrapy 的非分布式爬虫）

继承 CrawlSpider类的 Scrapy 爬虫

导入 CrawlSpider 类：from scrapy.spiders import CrawlSpider

(1) 生成爬虫

scrapy genspider -t crawl my_crawlspider "dmoz.org"

(2) 设置爬虫

vim my_crawlspider.py

# -*- coding: utf-8 -*-from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Ruleclass TestSpider(CrawlSpider):"""TestSpider 继承 CrawlSpider 爬虫类，也可以继承 scrapy.spider.Spider 类。但是如果继承 CrawlSpider，则可以定义一些规则(rule)来提供跟进 link 的方便的机制"""name = "test_spider"allowed_domains = ["dmoz.org"]start_urls = ["http://www.dmoz.org/"]links = LinkExtractor(restrict_css=('.top-cat', '.sub-cat', '.cat-item'))rules = [Rule(links, callback='parse_directory', follow=True),]def __init__(self):super(TestSpider, self).__init__()self.temp = Nonedef parse_directory(self, response):self.temp = Nonefor div in response.css('.title-and-desc'):data = {'name': div.css('.site-title::text').extract_first(),'description': div.css('.site-descr::text').extract_first(),                   'link': div.css('.a::attr(href)').extract(),}yield dataif __name__ == '__main__':from scrapy import cmdlinecmdline.execute('scrapy crawl test_spider'.split())pass

(3) 执行爬虫方法--- scrapy crawl my_crawlspider ( 与正常 scrapy 一样，无需 redis_key，比较鸡肋并不是真正的多机器爬虫)

示例：爬取豆瓣电影

# -*- coding: utf-8 -*-from scrapy.spiders import CrawlSpider, Rule
from scrapy.linkextractors import LinkExtractorclass DoubanSpider(CrawlSpider):name = 'douban'allowed_domains = ['movie.douban.com']start_urls = ['https://movie.douban.com/top250']custom_settings = {'DEFAULT_REQUEST_HEADERS': {'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,''*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',"Accept-Encoding": "gzip, deflate","Accept-Language": "zh-CN,zh;q=0.9","Connection": "keep-alive","Host": "movie.douban.com","Upgrade-Insecure-Requests": "1","User-Agent": 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'' (KHTML, like Gecko) Chrome/81.0.4044.113 Safari/537.36',},'CONCURRENT_REQUESTS': 10,'DOWNLOAD_DELAY': 0.01,'CONCURRENT_REQUESTS_PER_IP': 0,'CONCURRENT_REQUESTS_PER_DOMAIN': 10000,'FEED_EXPORT_ENCODING': 'utf-8'}# 定义一个爬取规则，从start_urls中的网页提取LinkExtractor上规定的所有链接，callback为对这些链接如何处理rules = (Rule(LinkExtractor(allow=(r'https://movie.douban.com/subject/\d+',)),callback='parse_item'),)def parse_item(self, response):print(self.name)data = dict()data['title'] = ''.join(response.xpath('//span[@property="v:itemreviewed"]/text()').extract())print(data)# yield dataif __name__ == '__main__':from scrapy import cmdlinecmdline.execute('scrapy crawl douban'.split())pass

类型二：( 基于 Scrapy-Redis 的分布式爬虫 )

基于 RedisSpider类的 Scrapy-Redis 分布式爬虫

(1) 生成爬虫--- scrapy genspider my_redisspider "dmoz.org"

(2) 设置爬虫--- vim my_redisspider.py

# -*- coding: utf-8 -*-# 变化 1: 从 scrapy_redis.spiders 中引入 RedisSpider
from scrapy_redis.spiders import RedisSpiderclass TestSpider(RedisSpider):  # 变化 2 : 爬虫类所继承的父类变为 RedisSpider类name = 'test_spider'# 变化 3 : 多了一个 redis_key，爬虫从这个 redis_key 里面取任务redis_key = "start_urls:test_spider"def __init__(self):super(TestSpider, self).__init__()def parse(self, response):# 直接将 name/url 存入 Redis数据库temp = {'name': response.css('.site-title::text').extract_first(),'url': response.url,}return tempif __name__ == '__main__':from scrapy import cmdlinecmdline.execute('scrapy crawl test_spider'.split())pass

(3) 执行爬虫方法

<1> 将项目代码复制到多台 slave 上 ( 可更改爬虫名 ) 并启动爬虫 --- scrapy runspider my_redisspider.py
<2> 往 redis 里面添加任务 --- lpush start_urls:test_spider http://www.dmoz.org/

类型三：（ Scrapy-Redis 的 RedisCrawlSpider 分布式爬虫）

基于 RedisCrawlSpider 类的 Scrapy-Redis 分布式爬虫

(1) 生成爬虫 --- scrapy genspider -t crawl my_rediscrawlspider "dmoz.org"
(2) 设置爬虫 --- vim my_rediscrawlspider.py

# -*- coding: utf-8 -*-from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule# 变化1: 从 scrapy_redis.spiders 中引入 RedisCrawlSpider
from scrapy_redis.spiders import RedisCrawlSpider# 变化2: 爬虫类所继承的父类变为 RedisCrawlSpider类
class TestSpider(RedisCrawlSpider):name = 'test_spider'# 变化3: 多了一个 redis_key，爬虫从这个 redis_key 里面取任务redis_key = 'start_urls:test_spider'rules = [Rule(LinkExtractor(), callback='parse_page', follow=True), ]def __init__(self):super(TestSpider, self).__init__()self.temp = Nonedef parse_page(self, response):self.temp = Nonedata = {'name': response.css('.site-title::text').extract_first(),'url': response.url,  # 直接将name/url存入Redis数据库}return dataif __name__ == '__main__':from scrapy import cmdlinecmdline.execute('scrapy crawl test_spider'.split())pass

(3) 执行爬虫方法

<1> 将项目代码复制到多台 slave上(可更改爬虫名)并启动爬虫 --- scrapy runspider my_rediscrawlspider.py
<2> 往 redis 里面添加任务 --- lpush start_urls:test_spider http://www.dmoz.org/

小结:

如果只想使用 Redis 的去重和保存功能，使 用类型一
如果写分布式，根据情况选择 类型二/类型三
如果写聚焦爬虫（全站式爬虫），选择 类型三

2.Scrapy-Rdis-project: youyuan

(Scrapy-Redis 分布式爬虫框架进阶1----有缘网:非分布式基于CrawlSpider的scrapy项目 )

1.创建项目

scrapy startproject youyuan

youyuan/
├── scrapy.cfg
└── youyuan
├── __init__.py
├── items.py
├── middlewares.py
├── pipelines.py
├── settings.py
└── spiders
├── __init__.py

└── yy.py

生成的项目目录截图

2.明确目标

vim items.py

improt scrapy                             class YouyuanItem(scrapy.Item):username = scrapy.Field()       # 用户名age = scrapy.Field()            # 年龄header_url = scrapy.Field()     # 头像地址image_url = scrapy.Field()      # 相册个图片地址content = scrapy.Field()        # 内心独白place_from = scrapy.Field()     # 籍贯education = scrapy.Field()      # 教育hobby = scrapy.Field()          # 爱好source_url = scrapy.Field()     # 个人主页source = scrapy.Field()         # 数据来源网站

3.制作爬虫

(1)生成爬虫--- scrapy genspider -t crawl yy "youyuan.com"
(2)设置爬虫--- vim yy.py

import scrapy
from scrapy.linkextractor import LinkExtractor
from scrapy.Spiders import CrawlSpider,Rule
from youyuan.items import YouyuanItem
import reclass YySpider(CrawlSpider):name = "yy"allowed_domains = ['youyuan.com']start_urls = ["http://www.youyuan.com/find/beijing/mm18-25/advance-0-0-0-0-0-0-0/p1/"]# 获取每个显示页的连接page_links = LinkExtractor(allow=(r'youyuan.com/find/beijing/mm18-25/advance-0-0-0-0-0-0-0/p\d+/')) # 获取个人主页person_links = LinkExtractor(allow =(r'youyuan.com/\d+-profile/'))rules = ( Rule(page_links),        # 没有callback/没有follow,默认follow=True继续跟进Rule(person_links,callback='parse_item'),   # 有callback/没有follow,默认follow=False不继续跟进)def parse_item(self,response):item = YouyuanItem()item['username'] = self.get_username()item['age'] = self.get_age()item['header_url'] = self.get_header_url()item['image_url'] = self.get_image_url()item['content'] = self.get_content()item['place_from'] = self.get_place_from()item['education'] = self.get_education()item['hobby'] = self.get_hobby()item['source_url'] = self.get_source_url()item['source'] = self.get_source()#-----所谓不再使用土枪土炮,看起来高大上的方法----------def get_username(self.response):username = response.xpath('//dl[@class="person_cen"]//div[@class="main"]/strong/text()').extract()if len(username):username = username[0]else:username = "Null"return username.strip()def get_age(self,response):age = response.xpath('//dl/[@class="person_cen"]//dd/p/text()').extract()if len(age):age = re.find_all(u"\d+岁",age[0])[0]else:age = "Null"return age.strip()def get_header_url(self,response):header_url = response.xpath('//dl[@class="person_cen"]//dt/img/@src').extract()if len (header_url):header_url = header_url[0]else:header_url = "Null"return header_url.strip()def get_image_url(self,response):image_url = response.xpath('//div[@class="ph_show"]/ul/li/a/img/@src').extract()if len(image_url):image_url = "|".join(image_url)  # 这样生成这种形式:xxxx|xxxx|xxxx|else:image_url = "Null"return image_urldef get_content(self,response):content = response.xpath('//div[@class="pre_data"]/ul/li[2]//ol[1]/span/text()').extract()if len(content):content = content[0]else:content = "Null"return content.strip()def get_place_from(self,response):place_from = response.xpath('//div[@class="pre_data"]/ul/li[2]//ol[2]/li[1]/span/text()').extract()if len(place_from):place_from = place_from[0]else:place_from = "Null"return place_from.strip()def get_education(self,response):education = response.xpath('//div[@class="pre_data"]/ul/li[3]//ol[2]//li[2]/span/text()').extract()if len(education):education = education[0]else:education = "Null"return education.strip()def get_hobby(self,response):hobby = response.xpath('//dl[@class="person_cen"]//ol/li/text()').extract()if len(hobby):hobby = ",".join(hobby).replace(" ","")else:hobby = "Null"return hobby.strip()

4.编写 item pipeline

vim pipelines.py

import json                             class YouyuanJsonPipeline(obeject):def __init__(self):self.f = open("youyuan.json","w")def process_item(self,item,spider):text = json.dumps(dict(item),ensure_ascii=False) + ",\n"self.f.write(item)def close_spider(self,spider):self.f.close()

5.启动上述 pipeline--- vim settings.py

vim settings.py
ITEM_PIPELINES = {"youyuan.pipelines.YouyuanJsonPipeline":300}

6.执行爬虫--- scrapy crawl yy

3.Scrapy-Redis-project: youyuan

(Scrapy-Redis分布式爬虫框架进阶2----有缘网:非分布式基于CrawlSpider的scrapy项目的数据存入本机Redis )

说明:仅仅实在上述scrapy项目的基础上进行settings.py文件的点滴修改,增加一个RedisPipeline而已
<----并不属于Scrapy-Redis分布式

youyaun/
├── scrapy.cfg
└── youyaun
├── __init__.py
├── items.py
├── middlewares.py
├── pipelines.py
├── settings.py <----仅对settings.py文件做部分添加修改
└── spiders
├── __init__.py
└── yy.py

5.settings.py添加部分信息--- 启用Scrapy-Redis的去重组件/调度器/取出请求方式,以及注册RedisPipeline组件(让数据存入Redis)

DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"  # 使用Scrapy-Redis的去重组件,不再使用scrapy的去重组件
SCHEDULER = "scrapy_redis.scheduler.Scheduler"              # 使用Scrapy-Redis的调度器,不再使用scrapy的调度器
# 使用Scrapy-Redis的从请求集合中取出请求的方式,三种方式择其一即可:
SCHEDULER_CLASS = "scrapy_redis.queue.SpiderPriorityQueue"  
# 分别按(1)请求的优先级/(2)队列FIFO/(3)栈FILO 取出请求
#SCHEDULER_CLASS = "scrapy_redis.queue.SpiderQueue"         
#SCHEDULER_CLASS = "scrapy_redis.queue.SpiderStack"
# 允许暂停,redis请求记录不会丢失(重启爬虫不会重头爬取已爬过的页面)
SCHEDULER_PERSIST = True                                    # 这两项是Redis连接设置,注释或不写默认将数据存放到本机的Redis中
#REDIS_HOST = "200.200.200.200"                             #REDIS_PORT = 6379              #----------注册RedisPipeline/自定义pipeline------------------
#注意:自定义pipeline的优先级需高于Redispipeline,因为RedisPipeline不会返回item,
#所以如果RedisPipeline优先级高于自定义pipeline,那么自定义pipeline无法获取到item
ITEM_PIPELINES = {"youyuan.pipelines.YouyuanJsonPipeline":300,        # 自定义pipeline视情况选择性注册(可选)# 将RedisPipeline注册到pipeline组件中(这样才能将数据存入Redis)"scrapy_redis.pipelines.RedisPipeline":400          }

6.执行爬虫--- scrapy crawl yy

# 注意: 在原始scrapy项目的基础上,在settings.py文件中添加上述几行设置,就可以将scrapy爬取的数据存放到本机redis中
# 注意: 上述要想成功保存到本机Redis,有两个前提:本机必须(pip install scrapy-redis) 和本机redis必须启动(redis-server /etc/redis.conf)

4.Scrapy-Rdis-project: youyuan

(Scrapy-Redis分布式爬虫框架进阶3----有缘网:非分布式基于CrawlSpider的scrapy项目 ----> 改写为:RedisCrawlSpider类的Scrapy-Redis分布式爬虫项目 )

说明:仅仅实在原始scrapy项目的基础上对settings.py/yy.py文件进行的点滴修改即可
youyaun/
├── scrapy.cfg
└── youyaun
├── __init__.py
├── items.py
├── middlewares.py
├── pipelines.py
├── settings.py <----对settings.py文件做部分添加修改
(使用Scrapy-Redis的去重组件/调度器/取出请求策略
/允许暂停/指明远程Redis主机,并注册RedisPipeline组件)
└── spiders
├── __init__.py
└── yy.py <----对爬虫文件进行部分修改(引入RedisCrawlSpider爬虫类/去掉allowed_domain/去掉start_urls/增加redis_key/改写init方法动态获取限制域)

1.修改设置文件--- vim settings.py

DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"  # 使用Scrapy-Redis的去重组件,不再使用scrapy的去重组件
SCHEDULER = "scrapy_redis.scheduler.Scheduler"              # 使用Scrapy-Redis的调度器,不再使用scrapy的调度器
# 使用Scrapy-Redis的从请求集合中取出请求的方式,三种方式择其一即可:
SCHEDULER_CLASS = "scrapy_redis.queue.SpiderPriorityQueue"  #SCHEDULER_CLASS = "scrapy_redis.queue.SpiderQueue"         # 分别按(1)请求的优先级/(2)队列FIFO/(3)栈FILO 取出请求
#SCHEDULER_CLASS = "scrapy_redis.queue.SpiderStack"
# 允许暂停,redis请求记录不会丢失(重启爬虫不会重头爬取已爬过的页面)
SCHEDULER_PERSIST = True                                    REDIS_HOST = "200.200.200.200"    # 这两项是Redis连接设置,如果注释或不写会默认将数据存放到本机的Redis中
REDIS_PORT = 6379      # 注意:master端的Redis需要允许远程连接--配置中注释掉bind 127.0.0.1         #----------注册RedisPipeline/自定义pipeline------------------
#注意:自定义pipeline的优先级需高于Redispipeline,因为RedisPipeline不会返回item,
#所以如果RedisPipeline优先级高于自定义pipeline,那么自定义pipeline无法获取到item
ITEM_PIPELINES = {"youyuan.pipelines.YouyuanJsonPipeline":300,  # 自定义pipeline视情况选择性注册(可选)"scrapy_redis.pipelines.RedisPipeline":400    # 将RedisPipeline注册到pipeline组件中(这样才能将数据存入Redis)
}

2. 修改爬虫文件--- vim yy.py

import scrapy
from scrapy linkextractor import LinkExtractor
from scrapy.Spiders import CrawlSpider,Rule
from youyuan.items import YouyuanItem
import re
from scrapy_redis.Spiders import RedisCrawlSpider    # 变化1:从scrapy_redis.Spiders中引入RedisCrawlSpiderclass YySpider(RedisCrawlSpider):          # 变化2:爬虫类所继承的父类变为RedisCrawlSpider类name = "yy"redis_key = "yyspider:start_urls"   # 变化3:多了一个对所有爬虫发号施令的redis_key,少了allowed_domain和start_urlsdef __init__(self,*args,**kwargs):                      # 变化4:重写__init__方法:动态获取限制域domain = kwargs.pop('domain','')self.allowed_domain = filter(None,domain,split(','))super(YySpider,self).__init__(*args,**kwargs)       # 注意super()里面的参数因爬虫类名不同而不同page_links = LinkExtractor(allow=(r'youyuan.com/find/beijing/mm18-25/advance-0-0-0-0-0-0-0/p\d+/')) person_links = LinkExtractor(allow =(r'youyuan.com/\d+-profile/'))  ..............      # 后面的代码都相同

3.爬虫的执行方式改变
<1>将项目代码复制到给台slave上(可更改爬虫名)并启动爬虫--- scrapy runspider yy.py
<2>在master端redis上发号施令--- lpush yyspider:start_urls http://www.youyuan.com/find/beijing/mm18-25/advance-0-0-0-0-0-0-0/p1/

5.Scrapy-Redis-project: sina2

(Scrapy-Redis分布式爬虫框架----新浪分类资讯:非分布式基于scrapy.Spider的scrapy项目 ----> 改写为:RedisSpider类的Scrapy-Redis分布式爬虫项目 ) 改写该项目注意:由于改写后数据存放到Redis,所以需要去掉"目录存储路径"相关的代码

1.创建项目--- scrapy startproject sina2
sina2/
├── scrapy.cfg
└── sina2
├── __init__.py
├── items.py
├── middlewares.py
├── pipelines.py
├── settings.py <----对settings.py文件做部分添加修改(使用Scrapy-Redis的去重组件/调度器/取出请求策略/
允许暂停/指明远程Redis主机,并注册RedisPipeline组件)
└── spiders
├── __init__.py
└── xinlang.py <----对爬虫文件进行部分修改(引入RedisCrawlSpider爬虫类/去掉allowed_domain
/去掉start_urls/增加redis_key/改写init方法动态获取限制域)

2.明确目标--- vim items.py
vim items.py
import scrapy

class SinaItem(scrapy.Item):parent_title = scrap.Field()       # 大类标题parent_url = scrap.Field()         # 大类URLsub_title = scrap.Field()          # 小类标题sub_url = scrap.Field()            # 小类URL#sub_filename = scrapy.Field()     # 小类目录存储路径(数据存放到Redis,不在需要这块代码)son_url = scrap.Field()            # 小类的子链接article_title = scrap.Field()      # 文章标题article_content = scrap.Field()    # 文章内容

3.制作爬虫

(1)生成爬虫--- scrapy genspider xinlang "sina.com.cn"
(2)设置爬虫--- vim xinlang.py

vim xinlang.pyimport scrapyimport osfrom sina.items import SinaItemimport sysfrom scrapy_redis.Spiders import RedisSpider                 ###<-----变化1:从scrapy_redis.Spiders中引入RedisSpiderreload(sys)sys.setdefaultencoding('utf-8')class XinlangSpider(RedisSpider):         ###<-----变化2:爬虫类所继承的父类变为RedisSpider类name = 'xinlang'              ##<-----变化3:多了一个对所有爬虫发号施令的redis_key,少了allowed_domain和start_urls                               redis_key = "xinlangspider:start_urls"                   def __init__(self,*args,**kwargs):                       ###<-----变化4:重写__init__方法:动态获取限制域domain = kwargs.pop('domain','')self.allowed_domain = filter(None,domain,split(','))super(XinlangSpider,self).__init__(*args,**kwargs)   # 注意super()里面的参数因爬虫类名不同而不同     def parse(self,response):items = []parent_title_list = response.xpath('//div[@id=\"tab01\"]/div/h3/a/text()').extract()parent_url_list = response.xpath('//div[@id=\"tab01\"]/div/h3/a/@href').extract()sub_title_list = response.xpath('//div[@id=\"tab01\"]/div/ul/li/a/text()').extract()sub_url_list = response.xpath('//div[@id=\"tab01\"]/div/ul/li/a/@href').extract()for i in range(0,len(parent_title_list)):   # 创建大类的存放目录(若存在则不创建,若不存在则重新创建)#parent_filename = "./Data/" + parent_title_list[i]   #if(not os.path.exists(parent_filename)):     # (数据存放到Redis,不在需要这块代码)#   os.makedirs(parent_filename) for j in range(0,len(sub_url_list)):           # 实例化SinaItem()并保存大类的URL和标题item = SinaItem()item['parent_title'] = parent_title_list[i]item['parent_url'] = parent_url_list[i]# 判断小类URL是否以大类URL开头(即判断小类是否属于大类)if_belong = sub_url_list[i].startwith(item['parent_url']) if (if_belong):                                             # 如果属于该大类,则判断小类存放目录是否存在,不存在则新建该目录#sub_filename = parent_filename + "/" + sub_title_list[j]   # (数据存放到Redis,不在需要这块代码)#if (not os.path.exists(sub_filename)):                     #   os.makedirs(sub_filename)# 保存小类的标题/URL/存放目录,并将目前所获取item信息追加到items列表中保存item['sub_title'] = sub_title_list[j]       item['sub_url'] = sub_url_list[j]item['sub_filename'] = sub_filenameitems.append(item)# 逐一取出子类的url,并附带上meta信息(即item),将其加入请求队列,使用second_parse()函数处理其返回的响应for item in items:              yield scrapy.Request(url=item['sub_url'],meta={'meta_1':item},callback=self.second_parse)def second_parse(self,response):meta_1 = response.meta['meta_1']                      # 将meta对应的item信息赋值给meta_1(即,meta_1 = item)son_url_list = response.xpath('//a/@href').extract()  # 匹配获取返回的孙类的URL列表items = []# 循环取出孙类URL判断其是否属于某个大类(以大类的URL开头)和是否是文章(以.shml结尾),# 如果属于则将该孙类URL保存起来for i in range(0,len(son_url_list)):    if_belong = son_url_list[i].endwith('.shtml') and sub_url_list[i].startwith(meta_1['parent_url'])if (if_belong):item = SinaItem()item['parent_title'] = meta_1['parent_title']item['parent_url'] = meta_1['parent_url']item['sub_title'] = meta_1['sub_title']item['sub_url'] = meta_1['sub_url']#item['sub_filename'] = meta_1['sub_filename']  # (数据存放到Redis,不在需要这块代码)item['son_url'] = son_url_list[i]items.append(item)# 逐一取出孙类的url,并附带上meta信息(即第二次的item),将其加入请求队列,使用third_parse()函数处理其返回的响应for item in items:              yield scrapy.Request(url=item['son_url'],meta={'meta_2':item},callback=self.third_parse)def third_parse(self,response):item = response.meta['meta_2']  # 将meta对应的(第二次获取更新的item信息)赋值给这里的item(即,item = item)article_content = ""            # 从孙类URL返回响应中匹配出文章标题和文章内容并保存进itemarticle_title_list = response.xpath('//hi[@id=\"main_title\"]/text()').extract()article_content_list = response.xpath('//div[@id=\"artibody\"]/p/text()').extract()for content_part in article_content_list:article_content += content_part               # 通过循环拼接成完整的文章内容item['article_title'] = article_title_list[0]item['article_content'] = article_contentyield item                                        # 将数据收集完整的item传递给pipeline处理

4.编写item pipelines

--- vim pipelines.py (忽略) # (数据存放到Redis,不再需要这块代码,不需要自定义pipeline保存数据到本地)

5.启用上述pipeline组件--- vim settings.py

DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"  # 使用Scrapy-Redis的去重组件,不再使用scrapy的去重组件
SCHEDULER = "scrapy_redis.scheduler.Scheduler"              # 使用Scrapy-Redis的调度器,不再使用scrapy的调度器
# 使用Scrapy-Redis的从请求集合中取出请求的方式,三种方式择其一即可:
SCHEDULER_CLASS = "scrapy_redis.queue.SpiderPriorityQueue"  #SCHEDULER_CLASS = "scrapy_redis.queue.SpiderQueue"    # 分别按(1)请求的优先级/(2)队列FIFO/(3)栈FILO 取出请求
#SCHEDULER_CLASS = "scrapy_redis.queue.SpiderStack"
SCHEDULER_PERSIST = True           # 允许暂停,redis请求记录不会丢失(重启爬虫不会重头爬取已爬过的页面)REDIS_HOST = "200.200.200.200"     # 这两项是Redis连接设置,如果注释或不写会默认将数据存放到本机的Redis中
REDIS_PORT = 6379                  # 注意:master端的Redis需要允许远程连接--配置中注释掉bind 127.0.0.1         # (数据存放到Redis,不再需要这块代码,不需要自定义pipeline保存数据到本地)
ITEM_PIPELINES = {  #"sina.pipelines.SinaSavePipeline":300,     # 将RedisPipeline注册到pipeline组件中(这样才能将数据存入Redis)"scrapy_redis.pipelines.RedisPipeline":400  
}

6.爬虫的执行方式改变

<1>将项目代码复制到给台slave上(可更改爬虫名)并启动爬虫--- scrapy runspider xinlang.py
<2>在master端redis上发号施令--- lpush yyspider:start_urls http://news.sina.com.cn/guide/

6.Scrapy-Rdis-project: youyuan

(Scrapy-Redis分布式爬虫框架----将Redis中数据持久化存储到MongoDB/MySQL中----> 将有缘网分布式爬取到Redis中的数据转存到MongoDB/MySQL中)

要将Scrapy-Redis项目爬取到Redis中的数据转存到Mongodb/MySQL中,只需要在项目一级目录下创建两个转存的脚本文件即可

有缘网Scrapy-Redis项目树形图
youyuan/
├── scrapy.cfg
├── process_item_for_mongodb.py <----项目一级目录下创建将Redis数据转存到mongodb的python脚本
├── process_item_for_mysql.py <----项目一级目录下创建将Redis数据转存到mysql的python脚本
└── youyuan
├── __init__.py
├── items.py
├── middlewares.py
├── pipelines.py
├── settings.py
└── spiders
├── __init__.py
└── yy.py

1.将Redis数据转存到mongodb中 ----- vim process_item_for_mongodb.py

#!/usr/bin/env python
#coding:utf-8import redis
import pymongo
import json
def process_item():rediscli = redis.Redis(host='200.200.200.200',port=6379,db=0)       # 创建Redis连接对象mongocli = pymongo.MongoClient(host='200.200.200.202',port=27017)   # 创建MongoDB连接对象db_name = mongocli['youyuan']     # 利用MongoDB连接对象在MongoDB中创建名为youyuan的数据库对象sheet_name = db_name['beijing18-24']  # 利用该数据库对象在youyuan数据库中创建名为beijing18-24的表对象count = 0 while True:# 使用循环通过redis连接对象的blpop()方法,不断取出redis中的数据(blpop即FIFO,rlpop即FILO)source,data = rediscli.blpop("yy:items")                        data = json.loads(data)   # 将取出的json字符串类型的数据转化为python类型的对象sheet_name.insert(data)   # 利用mongodb的表对象的insert()方法,向表中插入(刚才转化的python对象)count += 1print "已经成功从redis转移" + str(count) + "条数据到mongodb"if __name__ == "__main__":process_item()

注意:MongoDB中可以自动创建键,所以直接执行该脚本就可转移数据

2.将Redis数据转存到mysql中 ----- vim process_item_for_mysql.py

import redis
import MySQLdb
import json
def process_item():rediscli = redis.Redis(host='200.200.200.200',port=6379,db=0)     # 创建Redis连接对象# 创建MySQL连接对象mysqlcli = MySQLdb.connect(host='200.200.200.204',port 3306,db='youyuan',user='zhangsan',password='123456') count = 0while True:# 使用循环通过redis连接对象的blpop()方法,不断取出redis中的数据(blpop即FIFO,rlpop即FILO)source,data = rediscli.blpop('yy:items')           data = json.loads(data)      # 将取出的json字符串类型的数据转化为python类型的对象try:# 利用mysql连接对象创建cursor操作游标,并使用该操作游标向Mysql表中插入数据,数据通过python对象获取其值cursor = mysqlcli.cursor()                      cursor.execute("insert into beijing18-24 (username,age,spider,crawled)values(%s,%s,%s,%s)",[data['username'],data['age'],data['spider'],data['crawled']])mysqlcli.commit()                               # 插入完成后需提交事务cursor.close()                                  # 关闭操作游标count += 1print "已经成功从redis转移" + str(count) + "条数据到mongodb"except:passif __name__ == "__main__":process_item()

注意:MySQL中不能自动创建字段,所以在执行该脚本前,需要自行在数据库中创建好响应的数据库/表/字段,然后才能执行该脚本,转移数据到MySQL中