手把手教你搭建一个基于Java的分布式爬虫系统

手把手教你搭建一个基于Java的分布式爬虫系统

原创
作者：叶泳豪 2018-05-09 09:44:51

开发

后端

开发工具

分布式 在不用爬虫框架的情况下，我经过多方学习，尝试实现了一个分布式爬虫系统，并且可以将数据保存到不同地方，类似 MySQL、HBase 等。

专注于为中小企业提供成都网站制作、成都网站设计服务,电脑端+手机端+微信端的三站合一,更高效的管理,为中小企业南溪免费做网站提供优质的服务。我们立足成都，凝聚了一批互联网行业人才，有力地推动了上1000家企业的稳健成长，帮助中小企业通过网站建设实现规模扩充和转变。

【51CTO.com原创稿件】在不用爬虫框架的情况下，我经过多方学习，尝试实现了一个分布式爬虫系统，并且可以将数据保存到不同地方，类似 MySQL、HBase 等。

因为此系统基于面向接口的编码思想来开发，所以具有一定的扩展性，有兴趣的朋友直接看一下代码，就能理解其设计思想。

虽然代码目前来说很多地方还是比较紧耦合，但只要花些时间和精力，很多都是可抽取出来并且可配置化的。

因为时间的关系，我只写了京东和苏宁易购两个网站的爬虫，但是完全可以实现不同网站爬虫的随机调度，基于其代码结构，再写国美、天猫等的商品爬取，难度不大，但是估计需要花些时间和精力。

因为在解析网页的数据时，比如我在爬取苏宁易购商品的价格时，价格是异步获取的，并且其 API 是一长串的数字组合，我花了几个小时的时间才发现其规律，当然也承认，我的经验不足。

这个系统的设计，除了基本的数据爬取以外，更关注以下几个方面的问题：

  • 如何实现分布式？同一个程序打包后分发到不同的节点运行时，不影响整体的数据爬取。
  • 如何实现 URL 随机循环调度？核心是针对不同的***域名做随机。
  • 如何定时向 URL 仓库中添加种子 URL？达到不让爬虫系统停下来的目的。
  • 如何实现对爬虫节点程序的监控，并能够发邮件报警？
  • 如何实现一个随机 IP 代理库？目的跟第 2 点有点类似，都是为了反反爬虫。

下面会针对这个系统来做一个整体的基本介绍，我在代码中都有非常详细的注释，有兴趣的朋友可以参考一下代码，***我会给出一些我爬虫时的数据分析。

另外需要注意的是，这个爬虫系统是基于 Java 实现的，但是语言本身仍然不是最重要的，有兴趣的朋友可以尝试用 Python 实现。

分布式爬虫系统架构

整体系统架构如下：

从上面的架构可以看出，整个系统主要分为三个部分：

  • 爬虫系统
  • URL 调度系统
  • 监控报警系统

爬虫系统是用来爬取数据的，因为系统设计为分布式，因此，爬虫程序本身可以运行在不同的服务器节点上。

URL 调度系统核心在于 URL 仓库，所谓的 URL 仓库其实就是用 Redis 保存了需要爬取的 URL 列表，并且在我们的 URL 调度器中根据一定的策略来消费其中的 URL。从这个角度考虑，URL 仓库其实也是一个 URL 队列。

监控报警系统主要是对爬虫节点进行监控，虽然并行执行的爬虫节点中的某一个挂掉了对整体数据爬取本身没有影响（只是降低了爬虫的速度），但是我们还是希望能够主动接收到节点挂掉的通知，而不是被动地发现。

下面将针对以上三个方面并结合部分代码片段来对整个系统的设计思路做一些基本的介绍。

爬虫系统

爬虫系统是一个独立运行的进程，我们把我们的爬虫系统打包成 jar 包，然后分发到不同的节点上执行，这样并行爬取数据可以提高爬虫的效率。（说明：ZooKeeper 监控属于监控报警系统，URL 调度器属于 URL 调度系统）

随机 IP 代理器

加入随机 IP 代理主要是为了反反爬虫，因此如果有一个 IP 代理库，并且可以在构建 http 客户端时随机地使用不同的代理，那么对我们进行反反爬虫会有很大的帮助。

在系统中使用 IP 代理库，需要先在文本文件中添加可用的代理地址信息：

  
 
 
 
    
   
  
  
  
  1. # IPProxyRepository.txt 
    2. 
   
  
  
  
  2. 58.60.255.104:8118 
    3. 
   
  
  
  
  3. 219.135.164.245:3128 
    4. 
   
  
  
  
  4. 27.44.171.27:9999 
    5. 
   
  
  
  
  5. 219.135.164.245:3128 
    6. 
   
  
  
  
  6. 58.60.255.104:8118 
    7. 
   
  
  
  
  7. 58.252.6.165:9000 
    8. 
   
  
  
  
  8. ...... 
    9.

需要注意的是，上面的代理 IP 是我在西刺代理上拿到的一些代理 IP，不一定可用，建议是自己花钱购买一批代理 IP，这样可以节省很多时间和精力去寻找代理 IP。

然后在构建 http 客户端的工具类中，当***次使用工具类时，会把这些代理 IP 加载进内存中，加载到 Java 的一个 HashMap：

  
 
 
 
    
   
  
  
  
  1. // IP地址代理库Map 
    2. 
   
  
  
  
  2. private static Map IPProxyRepository = new HashMap<>(); 
    3. 
   
  
  
  
  3. private static String[] keysArray = null;   // keysArray是为了方便生成随机的代理对象 
    4. 
   
  
  
  
  4.  
    5. 
   
  
  
  
  5. /** 
    6. 
   
  
  
  
  6.      * 初次使用时使用静态代码块将IP代理库加载进set中 
    7. 
   
  
  
  
  7.      */ 
    8. 
   
  
  
  
  8. static { 
    9. 
   
  
  
  
  9.     InputStream in = HttpUtil.class.getClassLoader().getResourceAsStream("IPProxyRepository.txt");  // 加载包含代理IP的文本 
    10. 
   
  
  
  
  10.     // 构建缓冲流对象 
    11. 
   
  
  
  
  11.     InputStreamReader isr = new InputStreamReader(in); 
    12. 
   
  
  
  
  12.     BufferedReader bfr = new BufferedReader(isr); 
    13. 
   
  
  
  
  13.     String line = null; 
    14. 
   
  
  
  
  14.     try { 
    15. 
   
  
  
  
  15.         // 循环读每一行，添加进map中 
    16. 
   
  
  
  
  16.         while ((line = bfr.readLine()) != null) { 
    17. 
   
  
  
  
  17.             String[] split = line.split(":");   // 以:作为分隔符，即文本中的数据格式应为192.168.1.1:4893 
    18. 
   
  
  
  
  18.             String host = split[0]; 
    19. 
   
  
  
  
  19.             int port = Integer.valueOf(split[1]); 
    20. 
   
  
  
  
  20.             IPProxyRepository.put(host, port); 
    21. 
   
  
  
  
  21.         } 
    22. 
   
  
  
  
  22.         Set keys = IPProxyRepository.keySet(); 
    23. 
   
  
  
  
  23.         keysArray = keys.toArray(new String[keys.size()]);  // keysArray是为了方便生成随机的代理对象 
    24. 
   
  
  
  
  24.     } catch (IOException e) { 
    25. 
   
  
  
  
  25.         e.printStackTrace(); 
    26. 
   
  
  
  
  26.     } 
    27. 
   
  
  
  
  27.  
    28. 
   
  
  
  
  28. } 
    29.

之后，在每次构建 http 客户端时，都会先到 map 中看是否有代理 IP，有则使用，没有则不使用代理：

  
 
 
 
    
   
  
  
  
  1. CloseableHttpClient httpClient = null; 
    2. 
   
  
  
  
  2. HttpHost proxy = null; 
    3. 
   
  
  
  
  3. if (IPProxyRepository.size() > 0) {  // 如果ip代理地址库不为空，则设置代理 
    4. 
   
  
  
  
  4.     proxy = getRandomProxy(); 
    5. 
   
  
  
  
  5.     httpClient = HttpClients.custom().setProxy(proxy).build();  // 创建httpclient对象 
    6. 
   
  
  
  
  6. } else { 
    7. 
   
  
  
  
  7.     httpClient = HttpClients.custom().build();  // 创建httpclient对象 
    8. 
   
  
  
  
  8. } 
    9. 
   
  
  
  
  9. HttpGet request = new HttpGet(url); // 构建htttp get请求 
    10. 
   
  
  
  
  10. ...... 
    11.

随机代理对象则通过下面的方法生成：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.      * 随机返回一个代理对象 
    3. 
   
  
  
  
  3.      * 
    4. 
   
  
  
  
  4.      * @return 
    5. 
   
  
  
  
  5.      */ 
    6. 
   
  
  
  
  6. public static HttpHost getRandomProxy() { 
    7. 
   
  
  
  
  7.     // 随机获取host:port，并构建代理对象 
    8. 
   
  
  
  
  8.     Random random = new Random(); 
    9. 
   
  
  
  
  9.     String host = keysArray[random.nextInt(keysArray.length)]; 
    10. 
   
  
  
  
  10.     int port = IPProxyRepository.get(host); 
    11. 
   
  
  
  
  11.     HttpHost proxy = new HttpHost(host, port);  // 设置http代理 
    12. 
   
  
  
  
  12.     return proxy; 
    13. 
   
  
  
  
  13. } 
    14.

这样，通过上面的设计，基本就实现了随机 IP 代理器的功能，当然，其中还有很多可以完善的地方。

比如，当使用这个 IP 代理而请求失败时，是否可以把这一情况记录下来；当超过一定次数时，再将其从代理库中删除，同时生成日志供开发人员或运维人员参考，这是完全可以实现的，不过我就不做这一步功能了。

网页下载器

网页下载器就是用来下载网页中的数据，主要基于下面的接口开发：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * 网页数据下载 
    3. 
   
  
  
  
  3.  */ 
    4. 
   
  
  
  
  4. public interface IDownload { 
    5. 
   
  
  
  
  5.     /** 
    6. 
   
  
  
  
  6.      * 下载给定url的网页数据 
    7. 
   
  
  
  
  7.      * @param url 
    8. 
   
  
  
  
  8.      * @return 
    9. 
   
  
  
  
  9.      */ 
    10. 
   
  
  
  
  10.     public Page download(String url); 
    11. 
   
  
  
  
  11. } 
    12.

基于此，在系统中只实现了一个 http get 的下载器，但是也可以完成我们所需要的功能了：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * 数据下载实现类 
    3. 
   
  
  
  
  3.  */ 
    4. 
   
  
  
  
  4. public class HttpGetDownloadImpl implements IDownload { 
    5. 
   
  
  
  
  5.  
    6. 
   
  
  
  
  6.     @Override 
    7. 
   
  
  
  
  7.     public Page download(String url) { 
    8. 
   
  
  
  
  8.         Page page = new Page(); 
    9. 
   
  
  
  
  9.         String content = HttpUtil.getHttpContent(url);  // 获取网页数据 
    10. 
   
  
  
  
  10.         page.setUrl(url); 
    11. 
   
  
  
  
  11.         page.setContent(content); 
    12. 
   
  
  
  
  12.         return page; 
    13. 
   
  
  
  
  13.     } 
    14. 
   
  
  
  
  14. } 
    15.

网页解析器

网页解析器就是把下载的网页中我们感兴趣的数据解析出来，并保存到某个对象中，供数据存储器进一步处理以保存到不同的持久化仓库中，其基于下面的接口进行开发：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * 网页数据解析 
    3. 
   
  
  
  
  3.  */ 
    4. 
   
  
  
  
  4. public interface IParser { 
    5. 
   
  
  
  
  5.     public void parser(Page page); 
    6. 
   
  
  
  
  6. } 
    7.

网页解析器在整个系统的开发中也算是比较重头戏的一个组件，功能不复杂，主要是代码比较多，针对不同的商城不同的商品，对应的解析器可能就不一样了。

因此需要针对特别的商城的商品进行开发，因为很显然，京东用的网页模板跟苏宁易购的肯定不一样，天猫用的跟京东用的也肯定不一样。

所以这个完全是看自己的需要来进行开发了，只是说，在解析器开发的过程当中会发现有部分重复代码，这时就可以把这些代码抽象出来开发一个工具类了。

目前在系统中爬取的是京东和苏宁易购的手机商品数据，因此就写了这两个实现类：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * 解析京东商品的实现类 
    3. 
   
  
  
  
  3.  */ 
    4. 
   
  
  
  
  4. public class JDHtmlParserImpl implements IParser { 
    5. 
   
  
  
  
  5.     ...... 
    6. 
   
  
  
  
  6. } 
    7. 
   
  
  
  
  7.  
    8. 
   
  
  
  
  8. /** 
    9. 
   
  
  
  
  9.  * 苏宁易购网页解析 
    10. 
   
  
  
  
  10.  */ 
    11. 
   
  
  
  
  11. public class SNHtmlParserImpl implements IParser { 
    12. 
   
  
  
  
  12.     ...... 
    13. 
   
  
  
  
  13. } 
    14.

数据存储器

数据存储器主要是将网页解析器解析出来的数据对象保存到不同的表格，而对于本次爬取的手机商品，数据对象是下面一个 Page 对象：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * 网页对象，主要包含网页内容和商品数据 
    3. 
   
  
  
  
  3.  */ 
    4. 
   
  
  
  
  4. public class Page { 
    5. 
   
  
  
  
  5.     private String content;              // 网页内容 
    6. 
   
  
  
  
  6.  
    7. 
   
  
  
  
  7.     private String id;                    // 商品Id 
    8. 
   
  
  
  
  8.     private String source;               // 商品来源 
    9. 
   
  
  
  
  9.     private String brand;                // 商品品牌 
    10. 
   
  
  
  
  10.     private String title;                // 商品标题 
    11. 
   
  
  
  
  11.     private float price;                // 商品价格 
    12. 
   
  
  
  
  12.     private int commentCount;        // 商品评论数 
    13. 
   
  
  
  
  13.     private String url;                  // 商品地址 
    14. 
   
  
  
  
  14.     private String imgUrl;             // 商品图片地址 
    15. 
   
  
  
  
  15.     private String params;              // 商品规格参数 
    16. 
   
  
  
  
  16.  
    17. 
   
  
  
  
  17.     private List urls = new ArrayList<>();  // 解析列表页面时用来保存解析的商品url的容器 
    18. 
   
  
  
  
  18. } 
    19.

对应的，在 MySQL 中，表数据结构如下：

  
 
 
 
    
   
  
  
  
  1. -- ---------------------------- 
    2. 
   
  
  
  
  2. -- Table structure for phone 
    3. 
   
  
  
  
  3. -- ---------------------------- 
    4. 
   
  
  
  
  4. DROP TABLE IF EXISTS `phone`; 
    5. 
   
  
  
  
  5. CREATE TABLE `phone` ( 
    6. 
   
  
  
  
  6.   `id` varchar(30) CHARACTER SET armscii8 NOT NULL COMMENT '商品id', 
    7. 
   
  
  
  
  7.   `source` varchar(30) NOT NULL COMMENT '商品来源，如jd suning gome等', 
    8. 
   
  
  
  
  8.   `brand` varchar(30) DEFAULT NULL COMMENT '手机品牌', 
    9. 
   
  
  
  
  9.   `title` varchar(255) DEFAULT NULL COMMENT '商品页面的手机标题', 
    10. 
   
  
  
  
  10.   `price` float(10,2) DEFAULT NULL COMMENT '手机价格', 
    11. 
   
  
  
  
  11.   `comment_count` varchar(30) DEFAULT NULL COMMENT '手机评论', 
    12. 
   
  
  
  
  12.   `url` varchar(500) DEFAULT NULL COMMENT '手机详细信息地址', 
    13. 
   
  
  
  
  13.   `img_url` varchar(500) DEFAULT NULL COMMENT '图片地址', 
    14. 
   
  
  
  
  14.   `params` text COMMENT '手机参数，json格式存储', 
    15. 
   
  
  
  
  15.   PRIMARY KEY (`id`,`source`) 
    16. 
   
  
  
  
  16. ) ENGINE=InnoDB DEFAULT CHARSET=utf8; 
    17.

而在 HBase 中的表结构则为如下：

  
 
 
 
    
   
  
  
  
  1. ## cf1 存储 id source price comment brand url 
    2. 
   
  
  
  
  2. ## cf2 存储 title params imgUrl 
    3. 
   
  
  
  
  3. create 'phone', 'cf1', 'cf2' 
    4. 
   
  
  
  
  4.  
    5. 
   
  
  
  
  5. ## 在HBase shell中查看创建的表 
    6. 
   
  
  
  
  6. hbase(main):135:0> desc 'phone' 
    7. 
   
  
  
  
  7. Table phone is ENABLED                                                                                                 
    8. 
   
  
  
  
  8. phone                                                                                                                  
    9. 
   
  
  
  
  9. COLUMN FAMILIES DESCRIPTION                                                                                            
    10. 
   
  
  
  
  10. {NAME => 'cf1', BLOOMFILTER => 'ROW', VERSIONS => '1', IN_MEMORY => 'false', KEEP_DELETED_CELLS => 'FALSE', DATA_BLOCK 
    11. 
   
  
  
  
  11. _ENCODING => 'NONE', TTL => 'FOREVER', COMPRESSION => 'NONE', MIN_VERSIONS => '0', BLOCKCACHE => 'true', BLOCKSIZE =>  
    12. 
   
  
  
  
  12. '65536', REPLICATION_SCOPE => '0'}                                                                                     
    13. 
   
  
  
  
  13. {NAME => 'cf2', BLOOMFILTER => 'ROW', VERSIONS => '1', IN_MEMORY => 'false', KEEP_DELETED_CELLS => 'FALSE', DATA_BLOCK 
    14. 
   
  
  
  
  14. _ENCODING => 'NONE', TTL => 'FOREVER', COMPRESSION => 'NONE', MIN_VERSIONS => '0', BLOCKCACHE => 'true', BLOCKSIZE =>  
    15. 
   
  
  
  
  15. '65536', REPLICATION_SCOPE => '0'}                                                                                     
    16. 
   
  
  
  
  16. 2 row(s) in 0.0350 seconds 
    17.

即在 HBase 中建立了两个列族，分别为 cf1、cf2，其中 cf1 用来保存 id source price comment brand url 字段信息；cf2 用来保存 title params imgUrl 字段信息。

不同的数据存储用的是不同的实现类，但是其都是基于下面同一个接口开发的：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * 商品数据的存储 
    3. 
   
  
  
  
  3.  */ 
    4. 
   
  
  
  
  4. public interface IStore { 
    5. 
   
  
  
  
  5.     public void store(Page page); 
    6. 
   
  
  
  
  6. } 
    7.

然后基于此开发了 MySQL 的存储实现类、HBase 的存储实现类还有控制台的输出实现类，如 MySQL 的存储实现类，其实就是简单的数据插入语句：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * 使用dbc数据库连接池将数据写入mysql表中 
    3. 
   
  
  
  
  3.  */ 
    4. 
   
  
  
  
  4. public class MySQLStoreImpl implements IStore { 
    5. 
   
  
  
  
  5.     private QueryRunner queryRunner = new QueryRunner(DBCPUtil.getDataSource()); 
    6. 
   
  
  
  
  6.  
    7. 
   
  
  
  
  7.     @Override 
    8. 
   
  
  
  
  8.     public void store(Page page) { 
    9. 
   
  
  
  
  9.         String sql = "insert into phone(id, source, brand, title, price, comment_count, url, img_url, params) values(?, ?, ?, ?, ?, ?, ?, ?, ?)"; 
    10. 
   
  
  
  
  10.         try { 
    11. 
   
  
  
  
  11.             queryRunner.update(sql, page.getId(), 
    12. 
   
  
  
  
  12.                     page.getSource(), 
    13. 
   
  
  
  
  13.                     page.getBrand(), 
    14. 
   
  
  
  
  14.                     page.getTitle(), 
    15. 
   
  
  
  
  15.                     page.getPrice(), 
    16. 
   
  
  
  
  16.                     page.getCommentCount(), 
    17. 
   
  
  
  
  17.                     page.getUrl(), 
    18. 
   
  
  
  
  18.                     page.getImgUrl(), 
    19. 
   
  
  
  
  19.                     page.getParams()); 
    20. 
   
  
  
  
  20.         } catch (SQLException e) { 
    21. 
   
  
  
  
  21.             e.printStackTrace(); 
    22. 
   
  
  
  
  22.         } 
    23. 
   
  
  
  
  23.     } 
    24. 
   
  
  
  
  24. } 
    25.

而 HBase 的存储实现类，则是 HBase Java API 的常用插入语句代码：

  
 
 
 
    
   
  
  
  
  1. ...... 
    2. 
   
  
  
  
  2. // cf1:price 
    3. 
   
  
  
  
  3. Put pricePut = new Put(rowKey); 
    4. 
   
  
  
  
  4. // 必须要做是否为null判断，否则会有空指针异常 
    5. 
   
  
  
  
  5. pricePut.addColumn(cf1, "price".getBytes(), page.getPrice() != null ? String.valueOf(page.getPrice()).getBytes() : "".getBytes()); 
    6. 
   
  
  
  
  6. puts.add(pricePut); 
    7. 
   
  
  
  
  7. // cf1:comment 
    8. 
   
  
  
  
  8. Put commentPut = new Put(rowKey); 
    9. 
   
  
  
  
  9. commentPut.addColumn(cf1, "comment".getBytes(), page.getCommentCount() != null ? String.valueOf(page.getCommentCount()).getBytes() : "".getBytes()); 
    10. 
   
  
  
  
  10. puts.add(commentPut); 
    11. 
   
  
  
  
  11. // cf1:brand 
    12. 
   
  
  
  
  12. Put brandPut = new Put(rowKey); 
    13. 
   
  
  
  
  13. brandPut.addColumn(cf1, "brand".getBytes(), page.getBrand() != null ? page.getBrand().getBytes() : "".getBytes()); 
    14. 
   
  
  
  
  14. puts.add(brandPut); 
    15. 
   
  
  
  
  15. ...... 
    16.

当然，至于要将数据存储在哪个地方，在初始化爬虫程序时，是可以手动选择的：

  
 
 
 
    
   
  
  
  
  1. // 3.注入存储器 
    2. 
   
  
  
  
  2. iSpider.setStore(new HBaseStoreImpl()); 
    3.

目前还没有把代码写成可以同时存储在多个地方，按照目前代码的架构，要实现这一点也比较简单，修改一下相应代码就好了。

实际上，是可以先把数据保存到 MySQL 中，然后通过 Sqoop 导入到 HBase 中，详细操作可以参考我写的 Sqoop 文章。

仍然需要注意的是，如果确定需要将数据保存到 HBase 中，请保证你有可用的集群环境，并且需要将如下配置文档添加到 classpath 下：

  
 
 
 
    
   
  
  
  
  1. core-site.xml 
    2. 
   
  
  
  
  2. hbase-site.xml 
    3. 
   
  
  
  
  3. hdfs-site.xml 
    4.

对大数据感兴趣的同学可以折腾一下这一点，如果之前没有接触过的，直接使用 MySQL 存储就好了，只需要在初始化爬虫程序时注入 MySQL 存储器即可：

  
 
 
 
    
   
  
  
  
  1. // 3.注入存储器 
    2. 
   
  
  
  
  2. iSpider.setStore(new MySQLStoreImpl()); 
    3.

URL 调度系统

URL 调度系统是实现整个爬虫系统分布式的桥梁与关键，正是通过 URL 调度系统的使用，才使得整个爬虫系统可以较为高效（Redis 作为存储）随机地获取 URL，并实现整个系统的分布式。

URL 仓库

通过架构图可以看出，所谓的 URL 仓库不过是 Redis 仓库，即在我们的系统中使用 Redis 来保存 URL 地址列表。

正是这样，才能保证我们的程序实现分布式，只要保存了 URL 是唯一的，这样不管我们的爬虫程序有多少个，最终保存下来的数据都是只有唯一一份的，而不会重复。

同时 URL 仓库中的 URL 地址在获取时的策略是通过队列的方式来实现的，待会通过 URL 调度器的实现即可知道。

另外，在我们的 URL 仓库中，主要保存了下面的数据：

种子 URL 列表，Redis 的数据类型为 list

种子 URL 是持久化存储的，一定时间后，由 URL 定时器通过种子 URL 获取 URL，并将其注入到我们的爬虫程序需要使用的高优先级 URL 队列中。

这样就可以保证我们的爬虫程序可以源源不断地爬取数据而不需要中止程序的执行。

高优先级 URL 队列，Redis 的数据类型为 set

什么是高优先级 URL 队列？其实它就是用来保存列表 URL 的。那么什么是列表 URL 呢？

说白了就是一个列表中含有多个商品，以京东为例，我们打开一个手机列表：

该地址中包含的不是一个具体商品的 URL，而是包含了多个我们需要爬取的数据（手机商品）的列表。

通过对每个高级 URL 的解析，我们可以获取到非常多的具体商品 URL，而具体的商品 URL，就是低优先 URL，其会保存到低优先级 URL 队列中。

那么以这个系统为例，保存的数据类似如下：

  
 
 
 
    
   
  
  
  
  1. jd.com.higher 
    2. 
   
  
  
  
  2.     --https://list.jd.com/list.html?cat=9987,653,655&page=1 
    3. 
   
  
  
  
  3.     ...  
    4. 
   
  
  
  
  4. suning.com.higher 
    5. 
   
  
  
  
  5.     --https://list.suning.com/0-20006-0.html 
    6. 
   
  
  
  
  6.     ... 
    7.

低优先级 URL 队列，Redis 的数据类型为 set

低优先级 URL 其实就是具体某个商品的 URL，如下面一个手机商品：

通过下载该 URL 的数据，并对其进行解析，就能够获取到我们想要的数据。

那么以这个系统为例，保存的数据类似如下：

  
 
 
 
    
   
  
  
  
  1. jd.com.lower 
    2. 
   
  
  
  
  2.     --https://item.jd.com/23545806622.html 
    3. 
   
  
  
  
  3.     ... 
    4. 
   
  
  
  
  4. suning.com.lower 
    5. 
   
  
  
  
  5.     --https://product.suning.com/0000000000/690128156.html 
    6. 
   
  
  
  
  6.     ... 
    7.

URL 调度器

所谓 URL 调度器，就是 URL 仓库 Java 代码的调度策略，不过因为其核心在于调度，所以将其放到 URL 调度器中来进行说明，目前其调度基于以下接口开发：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * url 仓库 
    3. 
   
  
  
  
  3.  * 主要功能： 
    4. 
   
  
  
  
  4.  *      向仓库中添加url（高优先级的列表，低优先级的商品url） 
    5. 
   
  
  
  
  5.  *      从仓库中获取url（优先获取高优先级的url，如果没有，再获取低优先级的url） 
    6. 
   
  
  
  
  6.  * 
    7. 
   
  
  
  
  7.  */ 
    8. 
   
  
  
  
  8. public interface IRepository { 
    9. 
   
  
  
  
  9.  
    10. 
   
  
  
  
  10.     /** 
    11. 
   
  
  
  
  11.      * 获取url的方法 
    12. 
   
  
  
  
  12.      * 从仓库中获取url（优先获取高优先级的url，如果没有，再获取低优先级的url） 
    13. 
   
  
  
  
  13.      * @return 
    14. 
   
  
  
  
  14.      */ 
    15. 
   
  
  
  
  15.     public String poll(); 
    16. 
   
  
  
  
  16.  
    17. 
   
  
  
  
  17.     /** 
    18. 
   
  
  
  
  18.      * 向高优先级列表中添加商品列表url 
    19. 
   
  
  
  
  19.      * @param highUrl 
    20. 
   
  
  
  
  20.      */ 
    21. 
   
  
  
  
  21.     public void offerHigher(String highUrl); 
    22. 
   
  
  
  
  22.  
    23. 
   
  
  
  
  23.     /** 
    24. 
   
  
  
  
  24.      * 向低优先级列表中添加商品url 
    25. 
   
  
  
  
  25.      * @param lowUrl 
    26. 
   
  
  
  
  26.      */ 
    27. 
   
  
  
  
  27.     public void offerLower(String lowUrl); 
    28. 
   
  
  
  
  28.  
    29. 
   
  
  
  
  29. } 
    30.

其基于 Redis 作为 URL 仓库的实现如下：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * 基于Redis的全网爬虫，随机获取爬虫url： 
    3. 
   
  
  
  
  3.  * 
    4. 
   
  
  
  
  4.  * Redis中用来保存url的数据结构如下： 
    5. 
   
  
  
  
  5.  * 1.需要爬取的域名集合（存储数据类型为set，这个需要先在Redis中添加） 
    6. 
   
  
  
  
  6.  *      key 
    7. 
   
  
  
  
  7.  *          spider.website.domains 
    8. 
   
  
  
  
  8.  *      value(set) 
    9. 
   
  
  
  
  9.  *          jd.com  suning.com  gome.com 
    10. 
   
  
  
  
  10.  *      key由常量对象SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY 获得 
    11. 
   
  
  
  
  11.  * 2.各个域名所对应的高低优先url队列（存储数据类型为list，这个由爬虫程序解析种子url后动态添加） 
    12. 
   
  
  
  
  12.  *      key 
    13. 
   
  
  
  
  13.  *          jd.com.higher 
    14. 
   
  
  
  
  14.  *          jd.com.lower 
    15. 
   
  
  
  
  15.  *          suning.com.higher 
    16. 
   
  
  
  
  16.  *          suning.com.lower 
    17. 
   
  
  
  
  17.  *          gome.com.higher 
    18. 
   
  
  
  
  18.  *          gome.come.lower 
    19. 
   
  
  
  
  19.  *      value(list) 
    20. 
   
  
  
  
  20.  *          相对应需要解析的url列表 
    21. 
   
  
  
  
  21.  *      key由随机的域名 + 常量 SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX或者SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX获得 
    22. 
   
  
  
  
  22.  * 3.种子url列表 
    23. 
   
  
  
  
  23.  *      key 
    24. 
   
  
  
  
  24.  *          spider.seed.urls 
    25. 
   
  
  
  
  25.  *      value(list) 
    26. 
   
  
  
  
  26.  *          需要爬取的数据的种子url 
    27. 
   
  
  
  
  27.  *       key由常量SpiderConstants.SPIDER_SEED_URLS_KEY获得 
    28. 
   
  
  
  
  28.  * 
    29. 
   
  
  
  
  29.  *       种子url列表中的url会由url调度器定时向高低优先url队列中 
    30. 
   
  
  
  
  30.  */ 
    31. 
   
  
  
  
  31. public class RandomRedisRepositoryImpl implements IRepository { 
    32. 
   
  
  
  
  32.  
    33. 
   
  
  
  
  33.     /** 
    34. 
   
  
  
  
  34.      * 构造方法 
    35. 
   
  
  
  
  35.      */ 
    36. 
   
  
  
  
  36.     public RandomRedisRepositoryImpl() { 
    37. 
   
  
  
  
  37.         init(); 
    38. 
   
  
  
  
  38.     } 
    39. 
   
  
  
  
  39.  
    40. 
   
  
  
  
  40.     /** 
    41. 
   
  
  
  
  41.      * 初始化方法，初始化时，先将redis中存在的高低优先级url队列全部删除 
    42. 
   
  
  
  
  42.      * 否则上一次url队列中的url没有消耗完时，再停止启动跑下一次，就会导致url仓库中有重复的url 
    43. 
   
  
  
  
  43.      */ 
    44. 
   
  
  
  
  44.     public void init() { 
    45. 
   
  
  
  
  45.         Jedis jedis = JedisUtil.getJedis(); 
    46. 
   
  
  
  
  46.         Set domains = jedis.smembers(SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY); 
    47. 
   
  
  
  
  47.         String higherUrlKey; 
    48. 
   
  
  
  
  48.         String lowerUrlKey; 
    49. 
   
  
  
  
  49.         for(String domain : domains) { 
    50. 
   
  
  
  
  50.             higherUrlKey = domain + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX; 
    51. 
   
  
  
  
  51.             lowerUrlKey = domain + SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX; 
    52. 
   
  
  
  
  52.             jedis.del(higherUrlKey, lowerUrlKey); 
    53. 
   
  
  
  
  53.         } 
    54. 
   
  
  
  
  54.         JedisUtil.returnJedis(jedis); 
    55. 
   
  
  
  
  55.     } 
    56. 
   
  
  
  
  56.  
    57. 
   
  
  
  
  57.     /** 
    58. 
   
  
  
  
  58.      * 从队列中获取url，目前的策略是： 
    59. 
   
  
  
  
  59.      *      1.先从高优先级url队列中获取 
    60. 
   
  
  
  
  60.      *      2.再从低优先级url队列中获取 
    61. 
   
  
  
  
  61.      *  对应我们的实际场景，应该是先解析完列表url再解析商品url 
    62. 
   
  
  
  
  62.      *  但是需要注意的是，在分布式多线程的环境下，肯定是不能完全保证的，因为在某个时刻高优先级url队列中 
    63. 
   
  
  
  
  63.      *  的url消耗完了，但实际上程序还在解析下一个高优先级url，此时，其它线程去获取高优先级队列url肯定获取不到 
    64. 
   
  
  
  
  64.      *  这时就会去获取低优先级队列中的url，在实际考虑分析时，这点尤其需要注意 
    65. 
   
  
  
  
  65.      * @return 
    66. 
   
  
  
  
  66.      */ 
    67. 
   
  
  
  
  67.     @Override 
    68. 
   
  
  
  
  68.     public String poll() { 
    69. 
   
  
  
  
  69.         // 从set中随机获取一个***域名 
    70. 
   
  
  
  
  70.         Jedis jedis = JedisUtil.getJedis(); 
    71. 
   
  
  
  
  71.         String randomDomain = jedis.srandmember(SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY);    // jd.com 
    72. 
   
  
  
  
  72.         String key = randomDomain + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX;                // jd.com.higher 
    73. 
   
  
  
  
  73.         String url = jedis.lpop(key); 
    74. 
   
  
  
  
  74.         if(url == null) {   // 如果为null，则从低优先级中获取 
    75. 
   
  
  
  
  75.             key = randomDomain + SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX;    // jd.com.lower 
    76. 
   
  
  
  
  76.             url = jedis.lpop(key); 
    77. 
   
  
  
  
  77.         } 
    78. 
   
  
  
  
  78.         JedisUtil.returnJedis(jedis); 
    79. 
   
  
  
  
  79.         return url; 
    80. 
   
  
  
  
  80.     } 
    81. 
   
  
  
  
  81.  
    82. 
   
  
  
  
  82.     /** 
    83. 
   
  
  
  
  83.      * 向高优先级url队列中添加url 
    84. 
   
  
  
  
  84.      * @param highUrl 
    85. 
   
  
  
  
  85.      */ 
    86. 
   
  
  
  
  86.     @Override 
    87. 
   
  
  
  
  87.     public void offerHigher(String highUrl) { 
    88. 
   
  
  
  
  88.         offerUrl(highUrl, SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX); 
    89. 
   
  
  
  
  89.     } 
    90. 
   
  
  
  
  90.  
    91. 
   
  
  
  
  91.     /** 
    92. 
   
  
  
  
  92.      * 向低优先url队列中添加url 
    93. 
   
  
  
  
  93.      * @param lowUrl 
    94. 
   
  
  
  
  94.      */ 
    95. 
   
  
  
  
  95.     @Override 
    96. 
   
  
  
  
  96.     public void offerLower(String lowUrl) { 
    97. 
   
  
  
  
  97.         offerUrl(lowUrl, SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX); 
    98. 
   
  
  
  
  98.     } 
    99. 
   
  
  
  
  99.  
    100. 
   
  
  
  
  100.     /** 
    101. 
   
  
  
  
  101.      * 添加url的通用方法，通过offerHigher和offerLower抽象而来 
    102. 
   
  
  
  
  102.      * @param url   需要添加的url 
    103. 
   
  
  
  
  103.      * @param urlTypeSuffix  url类型后缀.higher或.lower 
    104. 
   
  
  
  
  104.      */ 
    105. 
   
  
  
  
  105.     public void offerUrl(String url, String urlTypeSuffix) { 
    106. 
   
  
  
  
  106.         Jedis jedis = JedisUtil.getJedis(); 
    107. 
   
  
  
  
  107.         String domain = SpiderUtil.getTopDomain(url);   // 获取url对应的***域名，如jd.com 
    108. 
   
  
  
  
  108.         String key = domain + urlTypeSuffix;            // 拼接url队列的key，如jd.com.higher 
    109. 
   
  
  
  
  109.         jedis.lpush(key, url);                          // 向url队列中添加url 
    110. 
   
  
  
  
  110.         JedisUtil.returnJedis(jedis); 
    111. 
   
  
  
  
  111.     } 
    112. 
   
  
  
  
  112. } 
    113.

通过代码分析也可以知道，其核心就在如何调度 URL 仓库（Redis）中的 URL。

URL 定时器

一段时间后，高优先级 URL 队列和低优先 URL 队列中的 URL 都会被消费完。

为了让程序可以继续爬取数据，同时减少人为的干预，可以预先在 Redis 中插入种子 URL，之后定时让 URL 定时器从种子 URL 中取出 URL 存放到高优先级 URL 队列中，以此达到程序定时不间断爬取数据的目的。

URL 消费完毕后，是否需要循环不断爬取数据根据个人业务需求而不同，因此这一步不是必需的，只是也提供了这样的操作。

因为事实上，我们需要爬取的数据也是每隔一段时间就会更新的，如果希望我们爬取的数据也跟着定时更新，那么这时定时器就有非常重要的作用了。

不过需要注意的是，一旦决定需要循环重复爬取数据，则在设计存储器实现时需要考虑重复数据的问题，即重复数据应该是更新操作。

目前在我设计的存储器不包括这个功能，有兴趣的朋友可以自己实现，只需要在插入数据前判断数据库中是否存在该数据即可。

另外需要注意的一点是，URL 定时器是一个独立的进程，需要单独启动。

定时器基于 Quartz 实现，下面是其 job 的代码：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * 每天定时从url仓库中获取种子url，添加进高优先级列表 
    3. 
   
  
  
  
  3.  */ 
    4. 
   
  
  
  
  4. public class UrlJob implements Job { 
    5. 
   
  
  
  
  5.  
    6. 
   
  
  
  
  6.     // log4j日志记录 
    7. 
   
  
  
  
  7.     private Logger logger = LoggerFactory.getLogger(UrlJob.class); 
    8. 
   
  
  
  
  8.  
    9. 
   
  
  
  
  9.     @Override 
    10. 
   
  
  
  
  10.     public void execute(JobExecutionContext context) throws JobExecutionException { 
    11. 
   
  
  
  
  11.         /** 
    12. 
   
  
  
  
  12.          * 1.从指定url种子仓库获取种子url 
    13. 
   
  
  
  
  13.          * 2.将种子url添加进高优先级列表 
    14. 
   
  
  
  
  14.          */ 
    15. 
   
  
  
  
  15.         Jedis jedis = JedisUtil.getJedis(); 
    16. 
   
  
  
  
  16.         Set seedUrls = jedis.smembers(SpiderConstants.SPIDER_SEED_URLS_KEY);  // spider.seed.urls Redis数据类型为set，防止重复添加种子url 
    17. 
   
  
  
  
  17.         for(String seedUrl : seedUrls) { 
    18. 
   
  
  
  
  18.             String domain = SpiderUtil.getTopDomain(seedUrl);   // 种子url的***域名 
    19. 
   
  
  
  
  19.             jedis.sadd(domain + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX, seedUrl); 
    20. 
   
  
  
  
  20.             logger.info("获取种子:{}", seedUrl); 
    21. 
   
  
  
  
  21.         } 
    22. 
   
  
  
  
  22.         JedisUtil.returnJedis(jedis); 
    23. 
   
  
  
  
  23. //        System.out.println("Scheduler Job Test..."); 
    24. 
   
  
  
  
  24.     } 
    25. 
   
  
  
  
  25.  
    26. 
   
  
  
  
  26. } 
    27.

调度器的实现如下：

  
 
 
 
    
   
  
  
  
  1. /** 
    2. 
   
  
  
  
  2.  * url定时调度器，定时向url对应仓库中存放种子url 
    3. 
   
  
  
  
  3.  * 
    4. 
   
  
  
  
  4.  * 业务规定：每天凌晨1点10分向仓库中存放种子url 
    5. 
   
  
  
  
  5.  */ 
    6. 
   
  
  
  
  6. public class UrlJobScheduler { 
    7. 
   
  
  
  
  7.  
    8. 
   
  
  
  
  8.     public UrlJobScheduler() { 
    9. 
   
  
  
  
  9.         init(); 
    10. 
   
  
  
  
  10.     } 
    11. 
   
  
  
  
  11.  
    12. 
   
  
  
  
  12.     /** 
    13. 
   
  
  
  
  13.      * 初始化调度器 
    14. 
   
  
  
  
  14.      */ 
    15. 
   
  
  
  
  15.     public void init() { 
    16. 
   
  
  
  
  16.         try { 
    17. 
   
  
  
  
  17.             Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler(); 
    18. 
   
  
  
  
  18.  
    19. 
   
  
  
  
  19.             // 如果没有以下start方法的执行，则是不会开启任务的调度 
    20. 
   
  
  
  
  20.               
    
                标题名称：手把手教你搭建一个基于Java的分布式爬虫系统
                
    
                链接URL：http://www.shufengxianlan.com/qtweb/news3/77903.html
            
    
            
    
                网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等
                                
    
            
    
                广告
                
            
    
            
    
                声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：
                创新互联

猜你还喜欢下面的内容

关键词优化知识

同城分类信息

成都商城网站开发　　 　眉山服务器租用　　 　小程序开发　　 　抖音短视频拍摄　　 　专业网站建设　　 　绵阳电信机房　　 　成都app开发　　 　广汉锦华　　 　晟和广告　　 　成都网站建设　　 　四川成都网站设计　　 　营销网站定位　　 　成都网站制作　　 　西信服务器托管　　 　攀枝花网站设计　　 　川西大数据中心　　 　四川鸿达　　 　泽君科技　　 　钢筋机械设备　　 　雅安主机托管