怎么样实现scrapy_redis爬虫分布式爬取功能？

import redis
import six
from scrapy.utils.misc import load_object
DEFAULT_REDIS_CLS = redis.StrictRedis
# Sane connection defaults.
DEFAULT_PARAMS = {
 'socket_timeout': 30,
 'socket_connect_timeout': 30,
 'retry_on_timeout': True,
}
# 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'])
 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)

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

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']

"""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)

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}

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

from scrapy import signals
from scrapy.exceptions import DontCloseSpider
from scrapy.spiders import Spider, CrawlSpider
from . import connection
class RedisMixin(object):
 """Mixin class to implement reading urls from a redis queue."""
 redis_key = None # If empty, uses default '<spider>:start_urls'.
 # Fetch this amount of start urls when idle.
 redis_batch_size = 100
 # 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 backwrads
  # compatibility.
  # XXX: Raise a deprecation warning.
  assert self.crawler, "crawler not set"
  crawler = self.crawler
 if not self.redis_key:
  self.redis_key = '%s:start_urls' % self.name
 self.log("Reading URLs from redis key '%s'" % self.redis_key)
 self.redis_batch_size = self.settings.getint(
  'REDIS_START_URLS_BATCH_SIZE',
  self.redis_batch_size,
 )
 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
  yield self.make_request_from_data(data)
  found += 1
 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):
 obj = super(RedisSpider, self).from_crawler(crawler)
 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):
 obj = super(RedisCrawlSpider, self).from_crawler(crawler)
 obj.setup_redis(crawler)
 return obj

{
 priority0:队列0
 priority1:队列2
 priority2:队列2
}