【學習目标】
- Scrapy-redis分布式的運作流程
- Scheduler與Scrapy自帶的Scheduler有什麼差別
- Duplication Filter作用
- 源碼自帶三種spider的使用
6. Scrapy-redis分布式元件
Scrapy 和 scrapy-redis的差別
Scrapy 是一個通用的爬蟲架構，但是不支援分布式，Scrapy-redis是為了更友善地實作Scrapy分布式爬取，而提供了一些以redis為基礎的元件(僅有元件)。
pip install scrapy-redis
Scrapy-redis提供了下面四種元件（components）：(四種元件意味着這四個子產品都要做相應的修改)
-
Scheduler
-
Duplication Filter
-
Item Pipeline
-
Base Spider
scrapy-redis架構
如上圖所⽰示，scrapy-redis在scrapy的架構上增加了redis，基于redis的特性拓展了如下元件：
Scheduler
Scheduler
Scrapy改造了python本來的collection.deque(雙向隊列)形成了自己的Scrapy queue(https://github.com/scrapy/queuelib/blob/master/queuelib/queue.py))，但是Scrapy多個spider不能共享待爬取隊列Scrapy queue， 即Scrapy本身不支援爬蟲分布式，scrapy-redis 的解決是把這個Scrapy queue換成redis資料庫（也是指redis隊列），從同一個redis-server存放要爬取的request，便能讓多個spider去同一個資料庫裡讀取。
Scrapy中跟“待爬隊列”直接相關的就是排程器
Scheduler
，它負責對新的request進行入列操作（加入Scrapy queue），取出下一個要爬取的request（從Scrapy queue中取出）等操作。它把待爬隊列按照優先級建立了一個字典結構，比如：
{ 優先級0 : 隊列0
優先級1 : 隊列1
優先級2 : 隊列2
}
然後根據request中的優先級，來決定該入哪個隊列，出列時則按優先級較小的優先出列。為了管理這個比較進階的隊列字典，Scheduler需要提供一系列的方法。但是原來的Scheduler已經無法使用，是以使用Scrapy-redis的scheduler元件。
Duplication Filter:
Scrapy中用集合實作這個request去重功能，Scrapy中把已經發送的request指紋放入到一個集合中，把下一個request的指紋拿到集合中比對，如果該指紋存在于集合中，說明這個request發送過了，如果沒有則繼續操作。這個核心的判重功能是這樣實作的：
def request_seen(self, request):
# self.request_figerprints就是一個指紋集合
fp = self.request_fingerprint(request)
# 這就是判重的核心操作
if fp in self.fingerprints:
return True
self.fingerprints.add(fp)
if self.file:
self.file.write(fp + os.linesep) 在scrapy-redis中去重是由
Duplication Filter
元件來實作的，它通過redis的set 不重複的特性，巧妙的實作了Duplication Filter去重。scrapy-redis排程器從引擎接受request，将request的指紋存⼊redis的set檢查是否重複，并将不重複的request push寫⼊redis的 request queue。
引擎請求request(Spider發出的）時，排程器從redis的request queue隊列⾥裡根據優先級pop 出⼀個request 傳回給引擎，引擎将此request發給spider處理。
Item Pipeline
：
Item Pipeline
：
引擎将(Spider傳回的)爬取到的Item給Item Pipeline，scrapy-redis 的Item Pipeline将爬取到的 Item 存⼊redis的 items queue。
修改過
Item Pipeline
可以很友善的根據 key 從 items queue 提取item，從⽽實作
items processes
叢集。
Base Spider:
不再使用scrapy原有的Spider類，重寫的
RedisSpider
繼承了Spider和RedisMixin這兩個類，RedisMixin是用來從redis讀取url的類。
當我們生成一個Spider繼承RedisSpider時，調用setup_redis函數，這個函數會去連接配接redis資料庫，然後會設定signals(信号)：
- 一個是當spider空閑時候的signal，會調用spider_idle函數，這個函數調用
schedule_next_request
- 一個是當抓到一個item時的signal，會調用item_scraped函數，這個函數會調用
schedule_next_request
6.1. 源碼分析參考：Connection
官方站點：https://github.com/rolando/scrapy-redis
scrapy-redis的官方文檔寫的比較簡潔，沒有提及其運作原理，是以如果想全面的了解分布式爬蟲的運作原理，還是得看scrapy-redis的源代碼才行。
scrapy-redis工程的主體還是是redis和scrapy兩個庫，工程本身實作的東西不是很多，這個工程就像膠水一樣，把這兩個插件粘結了起來。下面我們來看看，scrapy-redis的每一個源代碼檔案都實作了什麼功能，最後如何實作分布式的爬蟲系統：
connection.py
負責根據setting中配置執行個體化redis連接配接。被dupefilter和scheduler調用，總之涉及到redis存取的都要使用到這個子產品。
# 這裡引入了redis子產品，這個是redis-python庫的接口，用于通過python通路redis資料庫，
# 這個檔案主要是實作連接配接redis資料庫的功能，這些連接配接接口在其他檔案中經常被用到
import redis
import six
from scrapy.utils.misc import load_object
DEFAULT_REDIS_CLS = redis.StrictRedis
# 可以在settings檔案中配置套接字的逾時時間、等待時間等
# Sane connection defaults.
DEFAULT_PARAMS = {
'socket_timeout': 30,
'socket_connect_timeout': 30,
'retry_on_timeout': True,
}
# 要想連接配接到redis資料庫，和其他資料庫差不多，需要一個ip位址、端口号、使用者名密碼（可選）和一個整形的資料庫編号
# 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庫的Redis對象，可以直接用來進行資料操作的對象
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) connection.py
6.2. 源碼分析參考：Dupefilter
dupefilter.py
負責執行requst的去重，實作的很有技巧性，使用redis的set資料結構。但是注意scheduler并不使用其中用于在這個子產品中實作的dupefilter鍵做request的排程，而是使用queue.py子產品中實作的queue。
當request不重複時，将其存入到queue中，排程時将其彈出。
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 View Code
這個檔案看起來比較複雜，重寫了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在進入排程之前都要進行判重，如果重複就不需要參加排程，直接舍棄就好了，不然就是白白浪費資源。
6.3. 源碼分析參考：Picklecompat
picklecompat.py
"""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) picklecompat.py
這裡實作了loads和dumps兩個函數，其實就是實作了一個序列化器。
因為redis資料庫不能存儲複雜對象（key部分隻能是字元串，value部分隻能是字元串，字元串清單，字元串集合和hash），是以我們存啥都要先串行化成文本才行。
這裡使用的就是python的pickle子產品，一個相容py2和py3的串行化工具。這個serializer主要用于一會的scheduler存reuqest對象。
6.4. 源碼分析參考：Pipelines
pipelines.py
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} piplines.py
這是是用來實作分布式處理的作用。它将Item存儲在redis中以實作分布式處理。由于在這裡需要讀取配置，是以就用到了from_crawler()函數。
pipelines檔案實作了一個item pipieline類，和scrapy的item pipeline是同一個對象，通過從settings中拿到我們配置的
REDIS_ITEMS_KEY
作為key，把item串行化之後存入redis資料庫對應的value中（這個value可以看出是個list，我們的每個item是這個list中的一個結點），這個pipeline把提取出的item存起來，主要是為了友善我們延後處理資料。
6.5. 源碼分析參考：Queue
queue.py
該檔案實作了幾個容器類，可以看這些容器和redis互動頻繁，同時使用了我們上邊picklecompat中定義的序列化器。這個檔案實作的幾個容器大體相同，隻不過一個是隊列，一個是棧，一個是優先級隊列，這三個容器到時候會被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 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)
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)
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])
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)
__all__ = ['SpiderQueue', 'SpiderPriorityQueue', 'SpiderStack'] Queue
6.6. 源碼分析參考：Scheduler
scheduler.py
此擴充是對scrapy中自帶的scheduler的替代（在settings的SCHEDULER變量中指出），正是利用此擴充實作crawler的分布式排程。其利用的資料結構來自于queue中實作的資料結構。
scrapy-redis所實作的兩種分布式：爬蟲分布式以及item處理分布式就是由子產品scheduler和子產品pipelines實作。上述其它子產品作為為二者輔助的功能子產品
import importlib
import six
from scrapy.utils.misc import load_object
from . import connection
# 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 Scheduler.pu
這個檔案重寫了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進行爬取工作。
6.7. 源碼分析參考：Spider
spider.py
設計的這個spider從redis中讀取要爬的url,然後執行爬取，若爬取過程中傳回更多的url，那麼繼續進行直至所有的request完成。之後繼續從redis中讀取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
# Default batch size matches default concurrent requests setting.
DEFAULT_START_URLS_BATCH_SIZE = 16
DEFAULT_START_URLS_KEY = '%(name)s:start_urls'
class RedisMixin(object):
"""Mixin class to implement reading urls from a redis queue."""
# Per spider redis key, default to DEFAULT_START_URLS_KEY.
redis_key = None
# Fetch this amount of start urls when idle. Default to DEFAULT_START_URLS_BATCH_SIZE.
redis_batch_size = None
# Redis client instance.
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', DEFAULT_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:
self.redis_batch_size = settings.getint(
'REDIS_START_URLS_BATCH_SIZE', DEFAULT_START_URLS_BATCH_SIZE,
)
try:
self.redis_batch_size = int(self.redis_batch_size)
except (TypeError, ValueError):
raise ValueError("redis_batch_size must be an integer")
self.logger.info("Reading start URLs from redis key '%(redis_key)s' "
"(batch size: %(redis_batch_size)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')
fetch_one = self.server.spop if use_set else self.server.lpop
# XXX: Do we need to use a timeout here?
found = 0
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):
# By default, data is an URL.
if '://' in data:
return self.make_requests_from_url(data)
else:
self.logger.error("Unexpected URL from '%s': %r", self.redis_key, data)
def schedule_next_requests(self):
"""Schedules a request if available"""
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."""
@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."""
@classmethod
def from_crawler(self, crawler, *args, **kwargs):
obj = super(RedisCrawlSpider, self).from_crawler(crawler, *args, **kwargs)
obj.setup_redis(crawler)
return obj spider.py
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。
總結
最後總結一下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入口，然後再次重複上邊的工作。
7. Scrapy-redis實戰
從零搭建Redis-Scrapy分布式爬蟲
Scrapy-Redis分布式政策：
假設有三台電腦：Windows 10、Ubuntu 16.04、Windows 10，任意一台電腦都可以作為 Master端 或 Slaver端，比如：
l
Master
端
(核心伺服器) ：使用 Windows 10，搭建一個Redis資料庫，不負責爬取，隻負責url指紋判重、Request的配置設定，以及資料的存儲
l
Slaver
端
(爬蟲程式執行端) ：使用 Ubuntu 16.04、Windows 10，負責執行爬蟲程式，運作過程中送出新的Request給Master
- 首先Slaver端從Master端拿任務（Request、url）進行資料抓取，Slaver抓取資料的同時，産生新任務的Request便送出給 Master 處理；
- Master端隻有一個Redis資料庫，負責将未處理的Request去重和任務配置設定，将處理後的Request加入待爬隊列，并且存儲爬取的資料。
Scrapy-Redis預設使用的就是這種政策，我們實作起來很簡單，因為任務排程等工作Scrapy-Redis都已經幫我們做好了，我們隻需要繼承RedisSpider、指定redis_key就行了。
缺點是，Scrapy-Redis排程的任務是Request對象，裡面資訊量比較大（不僅包含url，還有callback函數、headers等資訊），可能導緻的結果就是會降低爬蟲速度、而且會占用Redis大量的存儲空間，是以如果要保證效率，那麼就需要一定硬體水準。
一、安裝Redis
安裝Redis：https://github.com/MSOpenTech/redis
安裝完成後，拷貝一份Redis安裝目錄下的redis.conf到任意目錄，建議儲存到：
/etc/redis/redis.conf
（Windows系統可以無需變動)
二、修改配置檔案 redis.conf
打開你的redis.conf配置檔案，示例:
l 非Windows系統:
sudo vi /etc/redis/redis.conf
l Windows系統：
C:\Intel\Redis\conf\redis.conf
- Master端redis.conf裡注釋
bind 127.0.0.1
- 在redis3.2之後，redis增加了protected-mode，在這個模式下，即使注釋掉了bind 127.0.0.1，再通路redis的時候還是報錯，如下：
有兩種解決方法：一種是設定protected-mode no，如下：
另一種是設定登入Ridis的密碼，如下：
三、測試Slave端遠端連接配接Master端
測試中，Master端Windows 10 的IP位址為：
192.168.199.108
- Master端按指定配置檔案啟動
redis-server
l 非Windows系統：
sudo redis-server /etc/redis/redis/conf
l Windows系統：
指令提示符(管理者)
模式下執行
redis-server.exe redis.windows.conf
讀取預設配置即可。
- Master端啟動本地
redis-cli
- slave端啟動
redis-cli -h 192.168.0.113 –a ye333222
注意：Slave端無需啟動
redis-server
，Master端啟動即可。隻要 Slave 端讀取到了 Master 端的 Redis 資料庫，則表示能夠連接配接成功，可以實施分布式。
四、Redis資料庫桌面管理工具
這裡推薦 Redis Desktop Manager，支援 Windows、Mac OS X、Linux 等平台：
下載下傳位址：https://redisdesktop.com/download
7.1. 源碼自帶項目說明
使用scrapy-redis的example來修改
先從github上拿到scrapy-redis的示例，然後将裡面的example-project目錄移到指定的位址：
# clone github scrapy-redis源碼檔案
git clone https://github.com/rolando/scrapy-redis.git # 直接拿官方的項目範例，改名為自己的項目用（針對懶癌患者)
mv scrapy-redis/example-project ~/scrapyredis-project 我們clone到的 scrapy-redis 源碼中有自帶一個example-project項目，這個項目包含3個spider，分别是dmoz, myspider_redis，mycrawler_redis。
一、dmoz (class DmozSpider(CrawlSpider))
這個爬蟲繼承的是CrawlSpider，它是用來說明Redis的持續性，當我們第一次運作dmoz爬蟲，然後Ctrl + C停掉之後，再運作dmoz爬蟲，之前的爬取記錄是保留在Redis裡的。
分析起來，其實這就是一個 scrapy-redis 版
CrawlSpider
類，需要設定Rule規則，以及callback不能寫parse()方法。
# 執行方式：scrapy crawl dmoz
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule
class DmozSpider(CrawlSpider):
"""Follow categories and extract links."""
name = 'dmoz'
allowed_domains = [' dmoztools.net']
start_urls = [' http://dmoztools.net/']
rules = [
Rule(LinkExtractor(
restrict_css=('.top-cat', '.sub-cat', '.cat-item')
), callback='parse_directory', follow=True),
]
def parse_directory(self, response):
for div in response.css('.title-and-desc'):
yield {
'name': div.css('.site-title::text').extract_first(),
'description': div.css('.site-descr::text').extract_first().strip(),
'link': div.css('a::attr(href)').extract_first(),
} 二、myspider_redis (class MySpider(RedisSpider))
這個爬蟲繼承了RedisSpider， 它能夠支援分布式的抓取，采用的是basic spider，需要寫parse函數。
其次就是不再有start_urls了，取而代之的是redis_key，scrapy-redis将key從Redis裡pop出來，成為請求的url位址。
from scrapy_redis.spiders import RedisSpider
class MySpider(RedisSpider):
"""Spider that reads urls from redis queue (myspider:start_urls)."""
name = 'myspider_redis'
# 注意redis-key的格式：
redis_key = 'myspider:start_urls'
# 可選：等效于allowd_domains()，__init__方法按規定格式寫，使用時隻需要修改super()裡的類名參數即可
def __init__(self, *args, **kwargs):
# Dynamically define the allowed domains list.
domain = kwargs.pop('domain', '')
self.allowed_domains = filter(None, domain.split(','))
# 修改這裡的類名為目前類名
super(MySpider, self).__init__(*args, **kwargs)
def parse(self, response):
return {
'name': response.css('title::text').extract_first(),
'url': response.url,
} 注意：
RedisSpider類 不需要寫
allowd_domains
和
start_urls
：
- scrapy-redis将從在構造方法
__init__()
allowd_domains
- 必須指定redis_key，即啟動爬蟲的指令，參考格式：redis_key = 'myspider:start_urls'
- 根據指定的格式，start_urls将在 Master端的 redis-cli 裡 lpush 到 Redis資料庫裡，RedisSpider 将在資料庫裡擷取start_urls。
執行方式：
- 通過runspider方法執行爬蟲的py檔案（也可以分次執行多條），爬蟲（們）将處于等待準備狀态：
scrapy runspider myspider_redis.py
- 在Master端的redis-cli輸入push指令，參考格式：
$redis > lpush myspider:start_urls http://dmoztools.net/
- Slaver端爬蟲擷取到請求，開始爬取。
三、mycrawler_redis (class MyCrawler(RedisCrawlSpider))
這個RedisCrawlSpider類爬蟲繼承了RedisCrawlSpider，能夠支援分布式的抓取。因為采用的是crawlSpider，是以需要遵守Rule規則，以及callback不能寫parse()方法。
同樣也不再有start_urls了，取而代之的是redis_key，scrapy-redis将key從Redis裡pop出來，成為請求的url位址。
from scrapy.spiders import Rule
from scrapy.linkextractors import LinkExtractor
from scrapy_redis.spiders import RedisCrawlSpider
class MyCrawler(RedisCrawlSpider):
"""Spider that reads urls from redis queue (myspider:start_urls)."""
name = 'mycrawler_redis'
redis_key = 'mycrawler:start_urls'
rules = (
# follow all links
Rule(LinkExtractor(), callback='parse_page', follow=True),
)
# __init__方法必須按規定寫，使用時隻需要修改super()裡的類名參數即可
def __init__(self, *args, **kwargs):
# Dynamically define the allowed domains list.
domain = kwargs.pop('domain', '')
self.allowed_domains = filter(None, domain.split(','))
# 修改這裡的類名為目前類名
super(MyCrawler, self).__init__(*args, **kwargs)
def parse_page(self, response):
return {
'name': response.css('title::text').extract_first(),
'url': response.url,
} 注意：
同樣的，RedisCrawlSpider類不需要寫
allowd_domains
和
start_urls
：
- scrapy-redis将從在構造方法
__init__()
allowd_domains
- 必須指定redis_key，即啟動爬蟲的指令，參考格式：
redis_key = 'mycrawl:start_urls'
- 根據指定的格式，
start_urls
執行方式：
- 通過runspider方法執行爬蟲的py檔案（也可以分次執行多條），爬蟲（們）将處于等待準備狀态：
scrapy runspider mycrawler_redis.py
- 在Master端的redis-cli輸入push指令，參考格式：
$redis > lpush mycrawler:start_urls http://dmoztools.net/
- 爬蟲擷取url，開始執行。
總結：
- 如果隻是用到Redis的去重和儲存功能，就選第一種；
- 如果要寫分布式，則根據情況，選擇第二種、第三種；
- 通常情況下，會選擇用第三種方式編寫深度聚焦爬蟲。
- Redis資料庫的使用
- RedisSpider類的使用
【重點總結】
- Redis資料庫的使用
- RedisSpider類的使用
轉載于:https://www.cnblogs.com/Mint-diary/p/9728435.html