redis批量获取hash key的数据-网站小程序定制开发-源码笔记-Erlo.vip
redis批量获取hash key的数据
时间： 17:18 & 阅读：14次 & 来源：php中文网
由于用户信息是缓存在redis hash类型中的需求是获取一个用户列表，比如每一页30个用户，想一次性获取这30个用户的用户信息 可是redis并没有批量获取hash key的方法！请问我该如何解决？是不是我的设计有问题？如用户 hash key 为 userinfo:1
user:info:2
user:info:3 user:info:4
....这些hash key 存储着用户信息想一次性获取 userinfo:1-30的hash值回复内容：由于用户信息是缓存在redis hash类型中的需求是获取一个用户列表，比如每一页30个用户，想一次性获取这30个用户的用户信息 可是redis并没有批量获取hash key的方法！请问我该如何解决？是不是我的设计有问题？如用户 hash key 为 userinfo:1
user:info:2
user:info:3 user:info:4
....这些hash key 存储着用户信息想一次性获取 userinfo:1-30的hash值用lua脚本循环eval &local rst={}; for i,v in pairs(KEYS) do rst[i]=redis.call('hgetall', v) return rst& 2 user:1 user:2127.0.0.1:6379& hgetall user:2
127.0.0.1:6379& hgetall user:1
127.0.0.1:6379& eval &local rst={}; for i,v in pairs(KEYS) do rst[i]=redis.call('hgetall', v) return rst& 2 user:1 user:2
1) 1) &name&
2) 1) &age&
4) &tom&用lua吧, 或者multi我的话就一个for循环，multi 是事务吧
还没有评论留言，赶紧来抢楼吧~~在redis中获取数据问题！ 谢谢~~~~~_百度知道
在redis中获取数据问题！ 谢谢~~~~~
在redis中放入所有的数据集合，在查询的时候根据条件怎么获取到相应的数据，有什么好的方法吗？
各位精通redis的大神么
我有更好的答案
redis是很纯粹的no sql，没办法。必须得条件查询的话，推荐是将要作为条件的列的值都拼接到redis的key中，然后使用程序+规则定制，使用程序拼出所有可能的主键，然后进行查询筛选，或者进行scan遍历。效率不高，没什么意义，还不如直接使用mysql。
采纳率：50%
为您推荐：
其他类似问题
换一换
回答问题，赢新手礼包
个人、企业类
违法有害信息,请在下方选择后提交
色情、暴力
我们会通过消息、邮箱等方式尽快将举报结果通知您。&>&&>& >Scrapy-redis爬虫散布式爬取的剖析和完成↓正文↓
Scrapy-redis爬虫散布式爬取的剖析和完成时间:来源:教程吧整集作者:未知编辑:php8本文已影响:2089人
Scrapy-redis爬虫散布式爬取的剖析和完成于搜集并整理至：python脚本 栏目。本站提示广大学习爱好者：（Scrapy-redis爬虫散布式爬取的剖析和完成）文章只能为提供参考，不一定能成为您想要的结果。
Scrapy是一个比拟好用的Python爬虫框架，你只需求编写几个组件就可以完成网页数据的爬取。但是当我们要爬取的页面十分多的时分，单个主机的处置才能就不能满足我们的需求了（无论是处置速度还是网络恳求的并发数），这时分散布式爬虫的优势就显现出来。
而Scrapy-Redis则是一个基于Redis的Scrapy散布式组件。它应用Redis对用于爬取的恳求(Requests)停止存储和调度(Schedule)，并对爬取发生的项目(items)存储以供后续处置运用。scrapy-redi重写了scrapy一些比拟关键的代码，将scrapy变成一个可以在多个主机上同时运转的散布式爬虫。
原生的Scrapy的架构是这样子的：
加上了Scrapy-Redis之后的架构变成了：
scrapy-redis的官方文档写的比拟简约，没有提及其运转原理，所以假如想片面的了解散布式爬虫的运转原理，还是得看scrapy-redis的源代码才行，不过scrapy-redis的源代码很少，也比拟好懂，很快就能看完。
scrapy-redis工程的主体还是是redis和scrapy两个库，工程自身完成的东西不是很多，这个工程好像胶水一样，把这两个插件粘结了起来。
scrapy-redis提供了哪些组件？
scrapy-redis所完成的两种散布式：爬虫散布式以及item处置散布式。辨别是由模块scheduler和模块pipelines完成。
connection.py
担任依据setting中配置实例化redis衔接。被dupefilter和scheduler调用，总之触及到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.
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)
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.
Extra parameters to be passed to the ``redis_cls`` class.
Redis client instance.
redis_cls = kwargs.pop('redis_cls', DEFAULT_REDIS_CLS)
url = kwargs.pop('url', None)
return redis_cls.from_url(url, **kwargs)
return redis_cls(**kwargs)
connect文件引入了redis模块，这个是redis-python库的接口，用于经过python访问redis数据库，可见，这个文件次要是完成衔接redis数据库的功用（前往的是redis库的Redis对象或许StrictRedis对象，这俩都是可以直接用来停止数据操作的对象）。这些衔接接口在其他文件中常常被用到。其中，我们可以看到，要想衔接到redis数据库，和其他数据库差不多，需求一个ip地址、端口号、用户名密码（可选）和一个整形的数据库编号，同时我们还可以在scrapy工程的setting文件中配置套接字的超时时间、等候时间等。
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.
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
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
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
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
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
这个文件看起来比拟复杂，重写了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在进入调度之前都要停止判重，假如反复就不需求参与调度，直接舍弃就好了，不然就是白白糜费资源。
其作用如dupefilter.py所述，但是这里完成了三种方式的queue：FIFO的SpiderQueue，SpiderPriorityQueue，以及LIFI的SpiderStack。默许运用的是第二种，这也就是呈现之前文章中所剖析状况的缘由（链接）。
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]
data = self.server.rpop(self.key)
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]
data = self.server.lpop(self.key)
return self._decode_request(data)
__all__ = ['SpiderQueue', 'SpiderPriorityQueue', 'SpiderStack']
该文件完成了几个容器类，可以看这些容器和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调度池，完成了散布式爬虫之间的一致调度。
picklecompat.py
"""A pickle wrapper module with protocol=-1 by default."""
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)
这里完成了loads和dumps两个函数，其实就是完成了一个serializer，由于redis数据库不能存储复杂对象（value局部只能是字符串，字符串列表，字符串集合和hash，key局部只能是字符串），所以我们存啥都要先串行化成文本才行。这里运用的就是python的pickle模块，一个兼容py2和py3的串行化工具。这个serializer次要用于一会的scheduler存reuqest对象，至于为什么不适用json格式，我也不是很懂，item pipeline的串行化默许用的就是json。
pipelines.py
这是是用来完成散布式处置的作用。它将Item存储在redis中以完成散布式处置。另外可以发现，异样是编写pipelines，在这里的编码完成不同于文章中所剖析的状况，由于在这里需求读取配置，所以就用到了from_crawler()函数。
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}
pipeline文件完成了一个item pipieline类，和scrapy的item pipeline是同一个对象，经过从settings中拿到我们配置的REDIS_ITEMS_KEY作为key，把item串行化之后存入redis数据库对应的value中（这个value可以看出出是个list，我们的每个item是这个list中的一个结点），这个pipeline把提取出的item存起来，次要是为了方便我们延后处置数据。
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)
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
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)(
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类，用来替代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停止爬取任务。
设计的这个spider从redis中读取要爬的url，然后执行爬取，若爬取进程中前往更多的url，那么持续停止直至一切的request完成。之后持续从redis中读取url，循环这个进程。
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:
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?
while found & self.redis_batch_size:
data = fetch_one(self.redis_key)
if not data:
# Queue empty.
yield self.make_request_from_data(data)
found += 1
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)
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
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入口，然后再次反复上边的任务。
为什么要提供这些组件？
我们先从scrapy的“待爬队列”和“Scheduler”动手：玩过爬虫的同窗都多多少少有些理解，在爬虫爬取进程当中有一个次要的数据构造是“待爬队列”，以及可以操作这个队列的调度器（也就是Scheduler）。scrapy官方文档对这二者的描绘不多，根本上没提。
scrapy运用什么样的数据构造来寄存待爬取的request呢？其实没用矮小上的数据构造，就是python自带的collection.deque（改造当时的），问题来了，该怎样让两个以上的Spider共用这个deque呢？
scrapy-redis提供了一个处理办法，把deque换成redis数据库，我们从同一个redis服务器寄存要爬取的request，这样就能让多个spider去同一个数据库里读取，这样散布式的次要问题就处理了嘛。
那么问题又来了，我们换了redis来寄存队列，哪scrapy就能直接散布式了么？。scrapy中跟“待爬队列”直接相关的就是调度器“Scheduler”，它担任对新的request停止出列操作（参加deque），取出下一个要爬取的request（从deque中取出）等操作。在scrapy中，Scheduler并不是直接就把deque拿来就粗犷的运用了，而且提供了一个比拟初级的组织办法，它把待爬队列依照优先级树立了一个字典构造，比方：
priority0:队列0
priority1:队列2
priority2:队列2
然后依据request中的priority属性，来决议该入哪个队列。而出列时，则按priority较小的优先出列。为了管理这个比拟初级的队列字典，Scheduler需求提供一系列的办法。你要是换了redis做队列，这个scrapy下的Scheduler就用不了，所以自己写一个吧。于是就呈现了scrapy-redis的公用scheduler。
那么既然运用了redis做次要数据构造，能不能把其他运用自带数据构造关键功用模块也换掉呢？ 在我们爬取进程当中，还有一个重要的功用模块，就是request去重。scrapy中是如何完成这个去重功用的呢？用集合~scrapy中把曾经发送的request指纹放入到一个集合中，把下一个request的指纹拿到集合中比对，假如该指纹存在于集合中，阐明这个request发送过了，假如没有则持续操作。
为了散布式，把这个集合也换掉吧，换了redis，照样也得把去重类给换了。于是就有了scrapy-redis的dupefilter。那么顺次类推，接上去的其他组件（Pipeline和Spider），我们也可以轻松的猜到，他们是为什么要被修正呢。
以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或许任务能带来一定的协助，假如有疑问大家可以留言交流。
Scrapy-redis爬虫散布式爬取的剖析和完成的相关资料介绍到这里，希望对您有所帮助！返回【&>&&>&】查看更多相关的资料！转载请保留本文连接地址：http://www.php8.org/python/93780.html
相关文章：
02-0902-0902-0902-0802-0802-0802-0502-0402-0302-03
相关标签：
最新推荐内容
热门点击阅读
Copyright&& &&
本站文章大部份源自于网络，其余为本站原创；如果本站文章侵犯了您的权益，请点击上方链接给我们留言。