From：https://cuiqingcai.com/6058.html
Scrapy-redis github：https://github.com/rmax/scrapy-redis scrapy-redis分布式爬虫框
集群版 Scrapy-Redis：https://github.com/thsheep/scrapy_redis_cluster scrapy-redis 和 scrapy 有什么区别？：https://www.zhihu.com/question/32302268 scrapy-redis使用以及剖
scrapy-redis
基于 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 = dupefilter
self.dqdir = self._dqdir(jobdir)
self.pqclass = pqclass
self.dqclass = dqclass
self.mqclass = mqclass
self.logunser = logunser
self.stats = stats
@classmethod
def from_crawler(cls, crawler):
'''
注意在 scrapy 中优先注意这个方法，此方法是一个钩子 用于访问当前爬虫的配置
'''
settings = crawler.settings
# 获取去重用的类 默认：scrapy.dupefilters.RFPDupeFilter
dupefilter_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方法 其实就是设置工作目录 和是否开启debug
dupefilter = dupefilter_cls.from_settings(settings)
# 获取优先级队列 类对象 默认：queuelib.pqueue.PriorityQueue
pqclass = load_object(settings['SCHEDULER_PRIORITY_QUEUE'])
# 获取磁盘队列 类对象（SCHEDULER使用磁盘存储 重启不会丢失）
dqclass = load_object(settings['SCHEDULER_DISK_QUEUE'])
# 获取内存队列 类对象（SCHEDULER使用内存存储 重启会丢失）
mqclass = load_object(settings['SCHEDULER_MEMORY_QUEUE'])
# 是否开启debug
logunser = 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) > 0
def 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是否已经请求过了 如果有会返回True
if 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 True
def 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返回给Engine
return request
def __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:
return
try:
reqd = request_to_dict(request, self.spider)
self.dqs.push(reqd, -request.priority)
except ValueError as e: # non serializable request
if 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 = False
self.stats.inc_value('scheduler/unserializable',
spider=self.spider)
return
else:
return True
def _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 q
def _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 = None
self.fingerprints = set()
self.logdupes = True
self.debug = debug
self.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)
@classmethod
def 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 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',
}
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 = 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_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 = 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 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.StrictRedis
The redis server instance.
key : str
Redis key Where to store fingerprints.
debug : bool, optional
Whether to log filtered 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.
"""
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)
@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
"""
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 == 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, reasnotallow=''):
"""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
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 pickle
def 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 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
@classmethod
def 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)
@classmethod
def 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 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.
"""
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_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
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
def _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 NotImplementedError
def push(self, request):
"""Push a request"""
raise NotImplementedError
def pop(self, timeout=0):
"""Pop a request"""
raise NotImplementedError
def 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 request
timeout not support in this queue class
"""
# use atomic range/remove using multi/exec
pipe = 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 six
from scrapy.utils.misc import load_object
from . import connection, defaults
# TODO: add SCRAPY_JOB support.
class Scheduler(object):
"""Redis-based scheduler
Settings
--------
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 : str
Scheduler redis key.
SCHEDULER_QUEUE_CLASS : str
Scheduler queue class.
SCHEDULER_DUPEFILTER_KEY : str
Scheduler dupefilter redis key.
SCHEDULER_DUPEFILTER_CLASS : str
Scheduler dupefilter class.
SCHEDULER_SERIALIZER : str
Scheduler 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 : Redis
The redis server instance.
persist : bool
Whether to flush requests when closing. Default is False.
flush_on_start : bool
Whether to flush requests on start. Default is False.
queue_key : str
Requests queue key.
queue_cls : str
Importable path to the queue class.
dupefilter_key : str
Duplicates filter key.
dupefilter_cls : str
Importable path to the dupefilter class.
idle_before_close : int
Timeout before giving up.
"""
if idle_before_close < 0:
raise TypeError("idle_before_close cannot be negative")
self.server = server
self.persist = persist
self.flush_on_start = flush_on_start
self.queue_key = queue_key
self.queue_cls = queue_cls
self.dupefilter_cls = dupefilter_cls
self.dupefilter_key = dupefilter_key
self.idle_before_close = idle_before_close
self.serializer = serializer
self.stats = None
def __len__(self):
return len(self.queue)
@classmethod
def 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)
@classmethod
def from_crawler(cls, crawler):
instance = cls.from_settings(crawler.settings)
# FIXME: for now, stats are only supported from this constructor
instance.stats = crawler.stats
return instance
def open(self, spider):
self.spider = spider
try:
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 crawl
if 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 False
if self.stats:
self.stats.inc_value('scheduler/enqueued/redis', spider=self.spider)
self.queue.push(request)
return True
def next_request(self):
block_pop_timeout = self.idle_before_close
request = self.queue.pop(block_pop_timeout)
if request and self.stats:
self.stats.inc_value('scheduler/dequeued/redis', spider=self.spider)
return request
def 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
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 redis
conn = redis.Redis(host='127.0.0.1',port=6379)
conn.lpush('start_urls:baidu','http://www.baidu.com')
添加 集合 形式 的 任务：
import redis
conn = 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:[email protected]:6379'
# 加上 decode_respnotallow=True，写入的键值对中的value为str类型，不加这个参数写入的则为字节类型。
r = redis.Redis.from_url(redis_url, decode_respnotallow=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, CrawlSpider
from . import connection, defaults
from .utils import bytes_to_str
class RedisMixin(object):
"""Mixin class to implement reading urls from a redis queue."""
redis_key = None
redis_batch_size = None
redis_encoding = None
# Redis client placeholder.
server = None
def 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:
return
if 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.settings
if 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 queue
crawler.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.
break
req = self.make_request_from_data(data)
if req:
yield req
found += 1
else:
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 to
provide your own message decoding.
Parameters
----------
data : bytes
Message 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 DontCloseSpider
class 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.
"""
@classmethod
def from_crawler(self, crawler, *args, **kwargs):
obj = super(RedisSpider, self).from_crawler(crawler, *args, **kwargs)
obj.setup_redis(crawler)
return obj
class 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.
"""
@classmethod
def 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 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.StrictRedis
The redis server instance.
key : str
Redis key Where to store fingerprints.
debug : bool, optional
Whether to log filtered 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.
"""
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)
@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
"""
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 == 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, reasnotallow=''):
"""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
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 scrapy
class 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 的方便的机制。
类型一：（基于 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, Rule
class 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 = None
def parse_directory(self, response):
self.temp = None
for 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 data
if __name__ == '__main__':
from scrapy import cmdline
cmdline.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 LinkExtractor
class 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 data
if __name__ == '__main__':
from scrapy import cmdline
cmdline.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 RedisSpider
class 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 temp
if __name__ == '__main__':
from scrapy import cmdline
cmdline.execute('scrapy crawl test_spider'.split())
pass
(3) 执行爬虫方法
- <1> 将项目代码复制到多台 slave 上 ( 可更改爬虫名 ) 并启动爬虫 --- scrapy runspider my_redisspider.py
- <2> 往 redis 里面添加任务 --- lpush start_urls:test_spider
类型三：（ 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 = None
def parse_page(self, response):
self.temp = None
data = {
'name': response.css('.site-title::text').extract_first(),
'url': response.url, # 直接将name/url存入Redis数据库
}
return data
if __name__ == '__main__':
from scrapy import cmdline
cmdline.execute('scrapy crawl test_spider'.split())
pass
(3) 执行爬虫方法
- <1> 将项目代码复制到多台 slave上(可更改爬虫名)并启动爬虫 --- scrapy runspider my_rediscrawlspider.py
- <2> 往 redis 里面添加任务 --- lpush start_urls:test_spider
小结:
- 如果只想使用 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 re
class 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_url
def 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中引入RedisCrawlSpider
class YySpider(RedisCrawlSpider): # 变化2:爬虫类所继承的父类变为RedisCrawlSpider类
name = "yy"
redis_key = "yyspider:start_urls" # 变化3:多了一个对所有爬虫发号施令的redis_key,少了allowed_domain和start_urls
def __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() # 大类URL
sub_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.py
import scrapy
import os
from sina.items import SinaItem
import sys
from scrapy_redis.Spiders import RedisSpider ###<-----变化1:从scrapy_redis.Spiders中引入RedisSpider
reload(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_filename
items.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返回响应中匹配出文章标题和文章内容并保存进item
article_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_content
yield 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-8
import 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 += 1
print "已经成功从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 = 0
while 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 += 1
print "已经成功从redis转移" + str(count) + "条数据到mongodb"
except:
pass
if __name__ == "__main__":
process_item()
注意:MySQL中不能自动创建字段,所以在执行该脚本前,需要自行在数据库中创建好响应的数据库/表/字段,然后才能执行该脚本,转移数据到MySQL中
标签：redis,self,request,scrapy,spider,Scrapy,源码,key From： https://blog.51cto.com/csnd/5956130