Scrapy框架学习（八）—-Scrapy-redis分布式爬虫学习

Scrapy-redis

分布式爬虫框架，是在

Scrapy

爬虫框架的基础上进行改进的，通过

Redis

来进行数据的缓存，可以在多台机器上运行爬虫程序。本文示例是在

CentOS

的虚拟机运行。

1、Redis安装

关于Redis的安装，网上有不少的文章，在配置Redis环境上也会有些问题，下面的2篇文章，详细的介绍了Redis的安装和问题，这里就直接搬运过来了。因为我的

Scrapy-Redis

的爬虫代码，是运行在

CentOS

中

CentOS 7下Redis的安装与配置

CentOS 7.0 安装Redis 3.2.1详细过程和使用常见问题

2、Redis分布式测试

我们采用3台机器，进行分布式爬虫爬取数据，在虚拟机上安装好

Redis

后，我们来进行

Redis

的分布式测试。

我们将server1作为

Master

机器（

ip:192.168.108.20

）。server2(

ip:192.168.108.21

)，server3(

ip:192.168.108.22

)作为2台机器为

Slaves

。

2.1 启动Master机器的Redis服务

找到redis的配置文件

redis.conf

，注释

bind 127.0.0.1

，保护模式

protected-mode

设置为

no

。

指定

redis.conf

启动

redis-server /redis/redis-stable/redis.conf
           

如图：

2.2 启动每台机器的Redis客户端redis-cli

Master机器：

Slaves机器，指定Master机器的ip

如果出现如下情况：

Could not connect to Redis at :: No route to host
           

需要对每台机器清除下防火墙配置：

sudo iptables -F
           

再次连接Master的Redis，如图：

如果机器连接不通，可能是防火墙的问题，可以试试：sudo iptables -F 清除防火墙配置

2.3 测试分布式Redis

3、scrapy-redis的爬虫代码

Scrapy-redis分布式爬虫，爬取的是：

http://sh.58.com/chuzu/

。

下面我们看下代码，基本和Scrapy的项目一样，就Spider的类改为Scrapy-redis中的类，settings.py中的配置不一样。

3.1 Item

定义实体类

class Redis58TestItem(scrapy.Item):
    # 标题
    title = scrapy.Field()
    # 房间
    room = scrapy.Field()
    # 区域
    zone = scrapy.Field()
    # 地址
    address = scrapy.Field()
    # 价格
    money = scrapy.Field()
    # 发布信息的类型，品牌公寓，经纪人，个人
    type = scrapy.Field()
           

3.2 spider

spider爬虫类实现

from scrapy.spiders import Rule
from scrapy_redis.spiders import RedisCrawlSpider
from scrapy.linkextractors import LinkExtractor
from redis58test.items import Redis58TestItem


class Redis58Spider(RedisCrawlSpider):# 继承的Spider是Scrapy-redis框架提供的
    # spider的唯一名称
    name = 'redis58spider_redis'
    # 开始爬取的url
    redis_key = 'redis58spider:start_urls'
    # 从页面需要提取的url 链接(link)
    links = LinkExtractor(allow="sh.58.com/chuzu/pn\d+")
    # 设置解析link的规则，callback是指解析link返回的响应数据的的方法
    rules = [Rule(link_extractor=links, callback="parseContent", follow=True)]

    def parseContent(self, response):
        """
        解析响应的数据，获取需要的数据字段
        :param response: 响应的数据
        :return:
        """
        # 根节点 //ul[@class="listUl"]/li[@logr]
        # title: .//div[@class="des"]/h2/a/text()
        # room: .//div[@class="des"]/p[@class="room"]/text()
        # zone: .//div[@class="des"]/p[@class="add"]/a[1]/text()
        # address: .//div[@class="des"]/p[@class="add"]/a[last()]/text()
        # money: .//div[@class="money"]/b/text()
        # type: # .//div[@class="des"]/p[last()]/@class     # 如果是add,room  .//div[@class="des"]/div[@class="jjr"]/@class

        for element in response.xpath('//ul[@class="listUl"]/li[@logr]'):
            title = element.xpath('.//div[@class="des"]/h2/a/text()')[].extract().strip()
            room = element.xpath('.//div[@class="des"]/p[@class="room"]')[].extract()
            zone = element.xpath('.//div[@class="des"]/p[@class="add"]/a[1]/text()')[].extract()
            address = element.xpath('.//div[@class="des"]/p[@class="add"]/a[last()]/text()')[].extract()
            money = element.xpath('.//div[@class="money"]/b/text()')[].extract()
            type = element.xpath('.//div[@class="des"]/p[last()]/@class')[].extract()
            if type == "add" or type == "room":
                type = element.xpath('.//div[@class="des"]/div[@class="jjr"]/@class')[].extract()

            item = Redis58TestItem()

            item['title'] = title
            item['room'] = room
            item['zone'] = zone
            item['address'] = address
            item['money'] = money
            item['type'] = type

            yield item
           

3.3 Pipeitem

实际上我们不需要处理Pipeitem的数据，因为Scrapy框架会将数据给redis去重。

class Redis58TestPipeline(object):
    def process_item(self, item, spider):
        return item
           

3.4 settings.py设置

配置Scrapy-Redis的信息

DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
SCHEDULER_PERSIST = True

# 使用什么队列调度
#SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.SpiderPriorityQueue"    # 优先级队列
#SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.SpiderQueue"  # 基本队列
#SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.SpiderStack"  # 栈

ITEM_PIPELINES = {
    'redis58test.pipelines.Redis58TestPipeline': ,
    'scrapy_redis.pipelines.RedisPipeline': ,
}


REDIS_HOST = '192.168.108.20'   # redis Master机器的ip
REDIS_PORT = 
           

4、执行爬虫

4.1 启动Master机器的redis-server

# 启动Master机器上的redis-server
redis-server /redis/redis-stable/redis.conf
           

4.2 执行爬虫程序

将Scrapy-redis项目的爬虫代码，放到slaves机器上。进入项目的spider目录，然后分别在slaves机器上启动scrapy-redis项目爬虫程序，不分先后。

4.3 Master机器的redis-cli

在Master机器的redis-cli里启动需要爬取的url。

5、将redis数据永久存储

爬虫程序爬取之后，我们将redis中的数据进行保存到mongodb数据库中。如下

import json
import redis
import pymongo


def main():
    # 指定Redis数据库信息
    rediscli = redis.StrictRedis(host='192.168.108.20', port=, db=)
    # 指定MongoDB数据库信息
    mongocli = pymongo.MongoClient(host='localhost', port=)

    # 创建数据库名
    db = mongocli['redis58test']
    # 创建表名
    sheet = db['redis58_sheet2']

    while True:
        # FIFO模式为 blpop，LIFO模式为 brpop，获取键值
        source, data = rediscli.blpop(["redis58spider_redis:items"])

        item = json.loads(data)
        sheet.insert(item)

        try:
            print(u"Processing: %(name)s <%(link)s>" % item)
        except KeyError:
            print(u"Error procesing: %r" % item)


if __name__ == '__main__':
    main()
           

数据如图：