Java 多线程爬虫及分布式爬虫架构探索

这是 Java 爬虫系列博文的第五篇，在上一篇 Java 爬虫服务器被屏蔽，不要慌，咱们换一台服务器 中，我们简单的聊反爬虫策略和反反爬虫方法，主要针对的是 IP 被封及其对应办法。前面几篇文章我们把爬虫相关的基本知识都讲的差不多啦。这一篇我们来聊一聊爬虫架构相关的内容。

前面几章内容我们的爬虫程序都是单线程，在我们调试爬虫程序的时候，单线程爬虫没什么问题，但是当我们在线上环境使用单线程爬虫程序去采集网页时，单线程就暴露出了两个致命的问题：

• 采集效率特别慢，单线程之间都是串行的，下一个执行动作需要等上一个执行完才能执行
• 对服务器的CUP等利用率不高，想想我们的服务器都是 8核16G，32G 的只跑一个线程会不会太浪费啦

线上环境不可能像我们本地测试一样，不在乎采集效率，只要能正确提取结果就行。在这个时间就是金钱的年代，不可能给你时间去慢慢的采集，所以单线程爬虫程序是行不通的，我们需要将单线程改成多线程的模式，来提升采集效率和提高计算机利用率。

多线程的爬虫程序设计比单线程就要复杂很多，但是与其他业务在高并发下要保证数据安全又不同，多线程爬虫在数据安全上到要求不是那么的高，因为每个页面都可以被看作是一个独立体。要做好多线程爬虫就必须做好两点：第一点就是统一的待采集 URL 维护，第二点就是 URL 的去重， 下面我们简单的来聊一聊这两点。

维护待采集的 URL

多线程爬虫程序就不能像单线程那样，每个线程独自维护这自己的待采集 URL，如果这样的话，那么每个线程采集的网页将是一样的，你这就不是多线程采集啦，你这是将一个页面采集的多次。基于这个原因我们就需要将待采集的 URL 统一维护，每个线程从统一 URL 维护处领取采集 URL ，完成采集任务，如果在页面上发现新的 URL 链接则添加到 统一 URL 维护的容器中。下面是几种适合用作统一 URL 维护的容器：

• JDK 的安全队列，例如 LinkedBlockingQueue
• 高性能的 NoSQL，比如 Redis、Mongodb
• MQ 消息中间件

URL 的去重

URL 的去重也是多线程采集的关键一步，因为如果不去重的话，那么我们将采集到大量重复的 URL，这样并没有提升我们的采集效率，比如一个分页的新闻列表，我们在采集第一页的时候可以得到 2、3、4、5 页的链接，在采集第二页的时候又会得到 1、3、4、5 页的链接，待采集的 URL 队列中将存在大量的列表页链接，这样就会重复采集甚至进入到一个死循环当中，所以就需要 URL 去重。URL 去重的方法就非常多啦，下面是几种常用的 URL 去重方式：

• 将 URL 保存到数据库进行去重，比如 redis、MongoDB
• 将 URL 放到哈希表中去重，例如 hashset
• 将 URL 经过 MD5 之后保存到哈希表中去重，相比于上面一种，能够节约空间
• 使用 布隆过滤器(Bloom Filter)去重，这种方式能够节约大量的空间，就是不那么准确。

关于多线程爬虫的两个核心知识点我们都知道啦，下面我画了一个简单的多线程爬虫架构图，如下图所示： 

多线程爬虫架构图

上面我们主要了解了多线程爬虫的架构设计，接下来我们不妨来试试 Java 多线程爬虫，我们以采集虎扑新闻为例来实战一下 Java 多线程爬虫，Java 多线程爬虫中设计到了 待采集 URL 的维护和 URL 去重，由于我们这里只是演示，所以我们就使用 JDK 内置的容器来完成，我们使用 LinkedBlockingQueue 作为待采集 URL 维护容器，HashSet 作为 URL 去重容器。下面是 Java 多线程爬虫核心代码，详细代码以上传 GitHub，地址在文末：

/** 
 * 多线程爬虫 
 */ 
public class ThreadCrawler implements Runnable { 
    // 采集的文章数 
    private final AtomicLong pageCount = new AtomicLong(0); 
    // 列表页链接正则表达式 
    public static final String URL_LIST = "https://voice.hupu.com/nba"; 
    protected Logger logger = LoggerFactory.getLogger(getClass()); 
    // 待采集的队列 
    LinkedBlockingQueue<String> taskQueue; 
    // 采集过的链接列表 
    HashSet<String> visited; 
    // 线程池 
    CountableThreadPool threadPool; 
    /** 
     * 
     * @param url 起始页 
     * @param threadNum 线程数 
     * @throws InterruptedException 
     */ 
    public ThreadCrawler(String url, int threadNum) throws InterruptedException { 
        this.taskQueue = new LinkedBlockingQueue<>(); 
        this.threadPool = new CountableThreadPool(threadNum); 
        this.visited = new HashSet<>(); 
        // 将起始页添加到待采集队列中 
        this.taskQueue.put(url); 
    } 
 
    @Override 
    public void run() { 
        logger.info("Spider started!"); 
        while (!Thread.currentThread().isInterrupted()) { 
            // 从队列中获取待采集 URL 
            final String request = taskQueue.poll(); 
            // 如果获取 request 为空，并且当前的线程采已经没有线程在运行 
            if (request == null) { 
                if (threadPool.getThreadAlive() == 0) { 
                    break; 
                } 
            } else { 
                // 执行采集任务 
                threadPool.execute(new Runnable() { 
                    @Override 
                    public void run() { 
                        try { 
                            processRequest(request); 
                        } catch (Exception e) { 
                            logger.error("process request " + request + " error", e); 
                        } finally { 
                            // 采集页面 +1 
                            pageCount.incrementAndGet(); 
                        } 
                    } 
                }); 
            } 
        } 
        threadPool.shutdown(); 
        logger.info("Spider closed! {} pages downloaded.", pageCount.get()); 
    } 
 
    /** 
     * 处理采集请求 
     * @param url 
     */ 
    protected void processRequest(String url) { 
        // 判断是否为列表页 
        if (url.matches(URL_LIST)) { 
            // 列表页解析出详情页链接添加到待采集URL队列中 
            processTaskQueue(url); 
        } else { 
            // 解析网页 
            processPage(url); 
        } 
    } 
    /** 
     * 处理链接采集 
     * 处理列表页，将 url 添加到队列中 
     * 
     * @param url 
     */ 
    protected void processTaskQueue(String url) { 
        try { 
            Document doc = Jsoup.connect(url).get(); 
            // 详情页链接 
            Elements elements = doc.select(" div.news-list > ul > li > div.list-hd > h4 > a"); 
            elements.stream().forEach((element -> { 
                String request = element.attr("href"); 
                // 判断该链接是否存在队列或者已采集的 set 中，不存在则添加到队列中 
                if (!visited.contains(request) && !taskQueue.contains(request)) { 
                    try { 
                        taskQueue.put(request); 
                    } catch (InterruptedException e) { 
                        e.printStackTrace(); 
                    } 
                } 
            })); 
            // 列表页链接 
            Elements list_urls = doc.select("div.voice-paging > a"); 
            list_urls.stream().forEach((element -> { 
                String request = element.absUrl("href"); 
                // 判断是否符合要提取的列表链接要求 
                if (request.matches(URL_LIST)) { 
                    // 判断该链接是否存在队列或者已采集的 set 中，不存在则添加到队列中 
                    if (!visited.contains(request) && !taskQueue.contains(request)) { 
                        try { 
                            taskQueue.put(request); 
                        } catch (InterruptedException e) { 
                            e.printStackTrace(); 
                        } 
                    } 
                } 
            })); 
 
        } catch (Exception e) { 
            e.printStackTrace(); 
        } 
    } 
    /** 
     * 解析页面 
     * 
     * @param url 
     */ 
    protected void processPage(String url) { 
        try { 
            Document doc = Jsoup.connect(url).get(); 
            String title = doc.select("body > div.hp-wrap > div.voice-main > div.artical-title > h1").first().ownText(); 
 
            System.out.println(Thread.currentThread().getName() + " 在 " + new Date() + " 采集了虎扑新闻 " + title); 
            // 将采集完的 url 存入到已经采集的 set 中 
            visited.add(url); 
 
        } catch (IOException e) { 
            e.printStackTrace(); 
        } 
    } 
 
    public static void main(String[] args) { 
 
        try { 
            new ThreadCrawler("https://voice.hupu.com/nba", 5).run(); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } 
    } 
} 

（编辑：甘孜站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!