1. 分布式爬蟲原理
Scrapy單機(jī)爬蟲有一個(gè)本地爬取隊(duì)列Queue，如果新的Request生成就會(huì)放到隊(duì)列里面，隨后Request被Scheduler調(diào)度，之后Request交給Downloader執(zhí)行。分布式爬蟲有多個(gè)Scheduler和多個(gè)Downloader，而爬取隊(duì)列始終為一個(gè)，也就是共享爬取隊(duì)列，這樣才能保證Scheduler從隊(duì)列里調(diào)度某個(gè)Request之后，其他的Scheduler不會(huì)重復(fù)調(diào)取此Request，就可以做到多個(gè)Scheduler同步爬取。
我們需要做的就是在多臺(tái)主機(jī)上同時(shí)運(yùn)行爬蟲任務(wù)，共享一個(gè)爬取隊(duì)列，各臺(tái)主機(jī)有自己的Scheduler和Downloader，這個(gè)共享的爬取隊(duì)列就是使用Redis來(lái)完成的。
2.scrapy-redis 源碼解析
Github地址：https://github.com/rmax/scrapy-redis/tree/master/src
各個(gè)組件功能介紹。
2.1 connection.py
負(fù)責(zé)根據(jù)setting中配置實(shí)例化redis連接。被dupefilter和scheduler調(diào)用，總之涉及到redis存取的都要使用到這個(gè)模塊。
# 這里引入了redis模塊，這個(gè)是redis-python庫(kù)的接口，用于通過(guò)python訪問(wèn)redis數(shù)據(jù)庫(kù)，
# 這個(gè)文件主要是實(shí)現(xiàn)連接redis數(shù)據(jù)庫(kù)的功能，這些連接接口在其他文件中經(jīng)常被用到
import redis
import six
from scrapy.utils.misc import load_object
DEFAULT_REDIS_CLS = redis.StrictRedis
# 可以在settings文件中配置套接字的超時(shí)時(shí)間、等待時(shí)間等
# Sane connection defaults.
DEFAULT_PARAMS = {
'socket_timeout': 30,
'socket_connect_timeout': 30,
'retry_on_timeout': True,
}
# 要想連接到redis數(shù)據(jù)庫(kù)，和其他數(shù)據(jù)庫(kù)差不多，需要一個(gè)ip地址、端口號(hào)、用戶名密碼（可選）和一個(gè)整形的數(shù)據(jù)庫(kù)編號(hào)
# Shortcut maps 'setting name' -> 'parmater name'.
SETTINGS_PARAMS_MAP = {
'REDIS_URL': 'url',
'REDIS_HOST': 'host',
'REDIS_PORT': 'port',
}
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
``DEFAULT_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_PARAMS : dict, optional
Additional client parameters.
"""
params = DEFAULT_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'])
# 返回的是redis庫(kù)的Redis對(duì)象，可以直接用來(lái)進(jìn)行數(shù)據(jù)操作的對(duì)象
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', DEFAULT_REDIS_CLS)
url = kwargs.pop('url', None)
if url:
return redis_cls.from_url(url, **kwargs)
else:
return redis_cls(**kwargs)
2.2 duperfilter.py
負(fù)責(zé)執(zhí)行Request的去重。Scrapy單機(jī)去重的過(guò)程就是利用集合元素的不重復(fù)性來(lái)實(shí)現(xiàn)，有一個(gè)request_fingerprint()方法就是Request指紋的方法，其內(nèi)部使用hashlib的sha1（）方法，計(jì)算的字段包括Request的Method，URL，Body，Headers這幾部分內(nèi)容，只要有一點(diǎn)不同那么計(jì)算的結(jié)果就不一樣，計(jì)算得到的是加密后的字符串，也就是指紋。每個(gè)Request都有一個(gè)獨(dú)有的指紋，判定字符串重復(fù)比判定Request對(duì)象是否重復(fù)要簡(jiǎn)單得多，所以指紋可以作為Request是否重復(fù)得依據(jù)。
Scrapy去重得實(shí)現(xiàn)：
def __init__(self):
self.fingerprints =set()
def request_seen(self, request):
fp = self.request_fingerprint(request)
if fp in self.fingerprints:
return True
self.fingerprints.add(fp)
檢測(cè)指紋是否存在于fingerprints變量中，該變量為一個(gè)集合，如果指紋存在就返回True，否則把這個(gè)指紋加入到集合中。
對(duì)于分布式爬蟲來(lái)說(shuō)，我們可以利用redis得集合作為指紋集合，那么這樣去重集合也是利用Redis共享的。每臺(tái)主機(jī)將新生成的Request指紋與集合對(duì)比，如果指紋已經(jīng)存在，就說(shuō)明Request是重復(fù)的。這樣，利用同樣的原理不同的存儲(chǔ)結(jié)構(gòu)就實(shí)現(xiàn)了分布式Request的去重。
duperfilter.py
import logging
import time
from scrapy.dupefilters import BaseDupeFilter
from scrapy.utils.request import request_fingerprint
from .connection import get_redis_from_settings
DEFAULT_DUPEFILTER_KEY = "dupefilter:%(timestamp)s"
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 = DEFAULT_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)
def close(self, reason=''):
"""Delete data on close. Called by Scrapy's scheduler.
Parameters
----------
reason : str, optional
"""
self.clear()
def clear(self):
"""Clears fingerprints data."""
self.server.delete(self.key)
def log(self, request, spider):
"""Logs given request.
Parameters
----------
request : scrapy.http.Request
spider : scrapy.spiders.Spider
"""
if self.debug:
msg = "Filtered duplicate request: %(request)s"
self.logger.debug(msg, {'request': request}, extra={'spider': spider})
elif self.logdupes:
msg = ("Filtered duplicate request %(request)s"
" - no more duplicates will be shown"
" (see DUPEFILTER_DEBUG to show all duplicates)")
msg = "Filtered duplicate request: %(request)s"
self.logger.debug(msg, {'request': request}, extra={'spider': spider})
self.logdupes = False
2.3 picklecompat.py
這里實(shí)現(xiàn)了loads和dumps兩個(gè)函數(shù)，其實(shí)就是實(shí)現(xiàn)了一個(gè)序列化器。
因?yàn)閞edis數(shù)據(jù)庫(kù)不能存儲(chǔ)復(fù)雜對(duì)象（key部分只能是字符串，value部分只能是字符串，字符串列表，字符串集合和hash），所以我們存儲(chǔ)之前都要先串行化成文本才行。
這里使用的就是python的pickle模塊，一個(gè)兼容py2和py3的串行化工具。
"""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)
2.4 pipelins.py
這是是用來(lái)實(shí)現(xiàn)分布式處理的作用。它將Item存儲(chǔ)在redis中以實(shí)現(xiàn)分布式處理。由于在這里需要讀取配置，所以就用到了from_crawler()函數(shù)。
pipelines文件實(shí)現(xiàn)了一個(gè)item pipieline類，和scrapy的item pipeline是同一個(gè)對(duì)象，通過(guò)從settings中拿到我們配置的REDIS_ITEMS_KEY作為key，把item串行化之后存入redis數(shù)據(jù)庫(kù)對(duì)應(yīng)的value中（這個(gè)value可以看出出是個(gè)list，我們的每個(gè)item是這個(gè)list中的一個(gè)結(jié)點(diǎn)），這個(gè)pipeline把提取出的item存起來(lái).
from scrapy.utils.misc import load_object
from scrapy.utils.serialize import ScrapyJSONEncoder
from twisted.internet.threads import deferToThread
from . import connection
default_serialize = ScrapyJSONEncoder().encode
class RedisPipeline(object):
"""Pushes serialized item into a redis list/queue"""
def __init__(self, server,
key='%(spider)s:items',
serialize_func=default_serialize):
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}
2.5 queue.py
它有三個(gè)隊(duì)列的實(shí)現(xiàn)，首先實(shí)現(xiàn)了一個(gè)父類Base，提供一些基本屬性和方法。
class Base(object):
"""Per-spider queue/stack base class"""
def __init__(self, server, spider, key, serializer=None):
"""Initialize per-spider redis queue.
Parameters:
server -- redis connection
spider -- spider instance
key -- key for this queue (e.g. "%(spider)s:queue")
"""
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)
_encode_request()跟_decode_request()方法實(shí)現(xiàn)了把一個(gè)Request對(duì)象存儲(chǔ)到數(shù)據(jù)庫(kù)的序列化操作。隊(duì)列Queue在調(diào)用oush方法將Request存入數(shù)據(jù)庫(kù)時(shí)調(diào)用encode方法進(jìn)行序列化，調(diào)用pop方法取出Request時(shí)會(huì)調(diào)用decode方法進(jìn)行反序列化。
在父類中， len(), push(), pop() 方法都是未實(shí)現(xiàn)的，所以必須實(shí)現(xiàn)一個(gè)子類來(lái)重寫這個(gè)三個(gè)方法。
有三個(gè)子類的實(shí)現(xiàn)：
Queue，PriorityQueue，Stack
- Queue：隊(duì)列，先進(jìn)先出
class SpiderQueue(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)
- PriorityQueue:優(yōu)先級(jí)隊(duì)列
class SpiderPriorityQueue(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])
這里使用的存儲(chǔ)結(jié)果時(shí)有序集合，滅個(gè)元素都可以設(shè)置一個(gè)分?jǐn)?shù)，這個(gè)分?jǐn)?shù)就代表優(yōu)先級(jí)。
此隊(duì)列是默認(rèn)使用的隊(duì)列
- Stack：棧 ，先進(jìn)后出
class SpiderStack(Base):
"""Per-spider stack"""
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)
2.6 scheduler.py
這個(gè)文件重寫了scheduler類，用來(lái)代替scrapy.core.scheduler的原有調(diào)度器。其實(shí)對(duì)原有調(diào)度器的邏輯沒有很大的改變，主要是使用了redis作為數(shù)據(jù)存儲(chǔ)的媒介，以達(dá)到各個(gè)爬蟲之間的統(tǒng)一調(diào)度。 scheduler負(fù)責(zé)調(diào)度各個(gè)spider的request請(qǐng)求，scheduler初始化時(shí)，通過(guò)settings文件讀取queue和dupefilters的類型，配置queue和dupefilters使用的key（一般就是spider name加上queue或者dupefilters，這樣對(duì)于同一種spider的不同實(shí)例，就會(huì)使用相同的數(shù)據(jù)塊了）。每當(dāng)一個(gè)request要被調(diào)度時(shí)，enqueue_request被調(diào)用，scheduler使用dupefilters來(lái)判斷這個(gè)url是否重復(fù)，如果不重復(fù)，就添加到queue的容器中（先進(jìn)先出，先進(jìn)后出和優(yōu)先級(jí)都可以，可以在settings中配置）。當(dāng)調(diào)度完成時(shí)，next_request被調(diào)用，scheduler就通過(guò)queue容器的接口，取出一個(gè)request，把他發(fā)送給相應(yīng)的spider，讓spider進(jìn)行爬取工作。
# TODO: add SCRAPY_JOB support.
class Scheduler(object):
"""Redis-based scheduler"""
def __init__(self, server,
persist=False,
flush_on_start=False,
queue_key='%(spider)s:requests',
queue_cls='scrapy_redis.queue.SpiderPriorityQueue',
dupefilter_key='%(spider)s:dupefilter',
dupefilter_cls='scrapy_redis.dupefilter.RFPDupeFilter',
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)
try:
self.df = load_object(self.dupefilter_cls)(
server=self.server,
key=self.dupefilter_key % {'spider': spider.name},
debug=spider.settings.getbool('DUPEFILTER_DEBUG'),
)
except TypeError as e:
raise ValueError("Failed to instantiate dupefilter class '%s': %s",
self.dupefilter_cls, e)
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
兩個(gè)核心的存取方法：enqueue_request()可以向隊(duì)列中添加Request，核心操作就是調(diào)用Queue的push操作，還有一些統(tǒng)計(jì)和日志操作。next_request()就是從隊(duì)列中取Request，調(diào)用pop操作，此時(shí)如果隊(duì)列中還有Request就會(huì)被取出來(lái)，繼續(xù)爬取，如果隊(duì)列為空，爬蟲就重新開始。
2.7 Spider
spider的改動(dòng)也不是很大，主要是通過(guò)connect接口，給spider綁定了spider_idle信號(hào)，spider初始化時(shí)，通過(guò)setup_redis函數(shù)初始化好和redis的連接，之后通過(guò)next_requests函數(shù)從redis中取出strat url，使用的key是settings中REDIS_START_URLS_AS_SET定義的（注意了這里的初始化url池和我們上邊的queue的url池不是一個(gè)東西，queue的池是用于調(diào)度的，初始化url池是存放入口url的，他們都存在redis中，但是使用不同的key來(lái)區(qū)分，就當(dāng)成是不同的表吧），spider使用少量的start url，可以發(fā)展出很多新的url，這些url會(huì)進(jìn)入scheduler進(jìn)行判重和調(diào)度。直到spider跑到調(diào)度池內(nèi)沒有url的時(shí)候，會(huì)觸發(fā)spider_idle信號(hào)，從而觸發(fā)spider的next_requests函數(shù)，再次從redis的start url池中讀取一些url。
3.總結(jié)
這個(gè)工程通過(guò)重寫scheduler和spider類，實(shí)現(xiàn)了調(diào)度、spider啟動(dòng)和redis的交互。實(shí)現(xiàn)新的dupefilter和queue類，達(dá)到了判重和調(diào)度容器和redis的交互，因?yàn)槊總€(gè)主機(jī)上的爬蟲進(jìn)程都訪問(wèn)同一個(gè)redis數(shù)據(jù)庫(kù)，所以調(diào)度和判重都統(tǒng)一進(jìn)行統(tǒng)一管理，達(dá)到了分布式爬蟲的目的。 當(dāng)spider被初始化時(shí)，同時(shí)會(huì)初始化一個(gè)對(duì)應(yīng)的scheduler對(duì)象，這個(gè)調(diào)度器對(duì)象通過(guò)讀取settings，配置好自己的調(diào)度容器queue和判重工具dupefilter。每當(dāng)一個(gè)spider產(chǎn)出一個(gè)request的時(shí)候，scrapy內(nèi)核會(huì)把這個(gè)reuqest遞交給這個(gè)spider對(duì)應(yīng)的scheduler對(duì)象進(jìn)行調(diào)度，scheduler對(duì)象通過(guò)訪問(wèn)redis對(duì)request進(jìn)行判重，如果不重復(fù)就把他添加進(jìn)redis中的調(diào)度池。當(dāng)調(diào)度條件滿足時(shí)，scheduler對(duì)象就從redis的調(diào)度池中取出一個(gè)request發(fā)送給spider，讓他爬取。當(dāng)spider爬取的所有暫時(shí)可用url之后，scheduler發(fā)現(xiàn)這個(gè)spider對(duì)應(yīng)的redis的調(diào)度池空了，于是觸發(fā)信號(hào)spider_idle，spider收到這個(gè)信號(hào)之后，直接連接redis讀取strart url池，拿去新的一批url入口，然后再次重復(fù)上邊的工作。
