摘要：各個(gè)爬蟲(chóng)的節(jié)點(diǎn)通過(guò)注冊(cè)到從而實(shí)現(xiàn)爬蟲(chóng)集群的管理。是一款基于等框架實(shí)現(xiàn)的通用爬蟲(chóng)框架。爬蟲(chóng)集群的監(jiān)控包含了和。用于實(shí)現(xiàn)爬蟲(chóng)的業(yè)務(wù)邏輯，可以添加到，由來(lái)管理各個(gè)的生命周期。

ZooKeeper 是一個(gè)開(kāi)源的分布式協(xié)調(diào)服務(wù)，ZooKeeper框架最初是在“Yahoo!"上構(gòu)建的，用于以簡(jiǎn)單而穩(wěn)健的方式訪問(wèn)他們的應(yīng)用程序。 后來(lái)，Apache ZooKeeper成為Hadoop，HBase和其他分布式框架使用的有組織服務(wù)的標(biāo)準(zhǔn)。 例如，Apache HBase使用ZooKeeper跟蹤分布式數(shù)據(jù)的狀態(tài)。ZooKeeper 的設(shè)計(jì)目標(biāo)是將那些復(fù)雜且容易出錯(cuò)的分布式一致性服務(wù)封裝起來(lái)，構(gòu)成一個(gè)高效可靠的原語(yǔ)集，并以一系列簡(jiǎn)單易用的接口提供給用戶(hù)使用。

ZooKeeper 通常用于：命名服務(wù)、配置管理、集群管理、分布式協(xié)調(diào)/通知、分布式鎖和分布式隊(duì)列等等。

各個(gè)爬蟲(chóng)的節(jié)點(diǎn)通過(guò)注冊(cè)到 ZooKeeper 從而實(shí)現(xiàn)爬蟲(chóng)集群的管理。NetDiscovery 正是借助了 ZooKeeper 的特性來(lái)監(jiān)控爬蟲(chóng)集群。

NetDiscovery 是一款基于 Vert.x、RxJava 2 等框架實(shí)現(xiàn)的通用爬蟲(chóng)框架。它包含了豐富的特性。

NetDiscovery 包含了 Spider 和 SpiderEngine。 Spider 用于實(shí)現(xiàn)爬蟲(chóng)的業(yè)務(wù)邏輯，Spider 可以添加到 SpiderEngine，由 SpiderEngine 來(lái)管理各個(gè) Spider 的生命周期。

但是 SpiderEngine 部署到每一個(gè)節(jié)點(diǎn)之后，SpiderEngine 如何進(jìn)行監(jiān)控和管理呢？

可以將 SpiderEngine 在運(yùn)行時(shí)，先注冊(cè)到 ZooKeeper。（需要事先在 ZooKeeper 集群創(chuàng)建 /netdiscovery 節(jié)點(diǎn)）

    /**
     * 啟動(dòng)SpiderEngine中所有的spider，讓每個(gè)爬蟲(chóng)并行運(yùn)行起來(lái)。
     *
     */
    public void run() {

        if (Preconditions.isNotBlank(spiders)) {

            registerZK();
            ......  
        }
    }

    /**
     * 將當(dāng)前 SpiderEngine 注冊(cè)到 zookeeper 指定的目錄 /netdiscovery 下
     */
    private void registerZK() {

        if (Preconditions.isNotBlank(zkStr) && useZk) {
            log.info("zkStr: {}", zkStr);

            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
            CuratorFramework client = CuratorFrameworkFactory.newClient(zkStr, retryPolicy);
            client.start();
            try {
                String ipAddr = InetAddress.getLocalHost().getHostAddress() + "-" + defaultHttpdPort + "-" + System.currentTimeMillis();
                String nowSpiderEngineZNode = "/netdiscovery/" + ipAddr;
                client.create().withMode(CreateMode.EPHEMERAL).forPath(nowSpiderEngineZNode,nowSpiderEngineZNode.getBytes());
            } catch (UnknownHostException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

另外，需要使用 NetDiscovery Monitor 的 CuratorManager 類(lèi)。 它借助 Zookeeper 的 Watcher 機(jī)制，監(jiān)聽(tīng)已經(jīng)注冊(cè)到 /netdiscovery 這個(gè)父 zNode 下的各個(gè)子 zNode ，也就是各個(gè) SpiderEngine。

Watcher機(jī)制是指 ZooKeeper 客戶(hù)端向 ZooKeeper 服務(wù)器注冊(cè) Watcher 的同時(shí)，會(huì)將 Watcher 對(duì)象存儲(chǔ)在客戶(hù)端的 WatchManager 中。ZooKeeper 服務(wù)器觸發(fā) Watcher 事件后，會(huì)向客戶(hù)端發(fā)送通知，客戶(hù)端線程從 WatchManager 中回調(diào) Watcher 執(zhí)行相應(yīng)的功能。

    /**
     * 當(dāng)前所監(jiān)控的父的 zNode 下若是子 zNode 發(fā)生了變化：新增，刪除，修改
     * 

     * 下述方法都會(huì)觸發(fā)執(zhí)行
     *
     * @param event
     */
    @Override
    public void process(WatchedEvent event) {

        List newZodeInfos = null;
        try {
            newZodeInfos = client.getChildren().usingWatcher(this).forPath("/netdiscovery");
            //根據(jù)初始化容器的長(zhǎng)度與最新的容器的長(zhǎng)度進(jìn)行比對(duì)，就可以推導(dǎo)出當(dāng)前 SpiderEngine 集群的狀態(tài)：新增，宕機(jī)/下線，變更...
            //哪個(gè)容器中元素多，就循環(huán)遍歷哪個(gè)容器。
            if (Preconditions.isNotBlank(newZodeInfos)) {
                if (newZodeInfos.size()>allZnodes.size()){
                    //明確顯示新增了哪個(gè) SpiderEngine 節(jié)點(diǎn)
                    for (String nowZNode:newZodeInfos) {
                        if (!allZnodes.contains(nowZNode)){
                            log.info("新增 SpiderEngine 節(jié)點(diǎn){}", nowZNode);
                        }
                    }
                }else if (newZodeInfos.size()// 宕機(jī)/下線
                    // 明確顯示哪個(gè) SpiderEngine 節(jié)點(diǎn)宕機(jī)/下線了
                    for (String initZNode : allZnodes) {
                        if (!newZodeInfos.contains(initZNode)) {
                            log.info("SpiderEngine 節(jié)點(diǎn)【{}】下線了！", initZNode);

                            // 如果有下線的處理，則處理(例如發(fā)郵件、短信等)
                            if (serverOfflineProcess!=null) {
                                serverOfflineProcess.process();
                            }
                        }
                    }
                }else {
                    // SpiderEngine 集群正常運(yùn)行;
                    // 宕機(jī)/下線了，當(dāng)時(shí)馬上重啟了，總的爬蟲(chóng)未發(fā)生變化
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        allZnodes = newZodeInfos;
    }

所以需要多帶帶運(yùn)行一個(gè)進(jìn)程，例如：

public class TestCuratorManager {

    public static void main(String[] args) {

        CuratorManager curatorManager = new CuratorManager();
        curatorManager.start();
    }
}

下圖反映了 ZooKeeper 如何監(jiān)控 SpiderEngine 集群。

總結(jié)

爬蟲(chóng)框架 github 地址：github.com/fengzhizi71…

本文介紹了如何使用 ZooKeeper 來(lái)監(jiān)控爬蟲(chóng)的集群。未來(lái)，NetDiscovery 還會(huì)增加更為通用的功能。

Java與Android技術(shù)棧：每周更新推送原創(chuàng)技術(shù)文章，歡迎掃描下方的公眾號(hào)二維碼并關(guān)注，期待與您的共同成長(zhǎng)和進(jìn)步。