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

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

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

各個爬蟲的節(jié)點通過注冊到 ZooKeeper 從而實現(xiàn)爬蟲集群的管理。NetDiscovery 正是借助了 ZooKeeper 的特性來監(jiān)控爬蟲集群。

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

NetDiscovery 包含了 Spider 和 SpiderEngine。 Spider 用于實現(xiàn)爬蟲的業(yè)務邏輯，Spider 可以添加到 SpiderEngine，由 SpiderEngine 來管理各個 Spider 的生命周期。

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

可以將 SpiderEngine 在運行時，先注冊到 ZooKeeper。（需要事先在 ZooKeeper 集群創(chuàng)建 /netdiscovery 節(jié)點）

    /**
     * 啟動SpiderEngine中所有的spider，讓每個爬蟲并行運行起來。
     *
     */
    public void run() {

        if (Preconditions.isNotBlank(spiders)) {

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

    /**
     * 將當前 SpiderEngine 注冊到 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 類。 它借助 Zookeeper 的 Watcher 機制，監(jiān)聽已經(jīng)注冊到 /netdiscovery 這個父 zNode 下的各個子 zNode ，也就是各個 SpiderEngine。

Watcher機制是指 ZooKeeper 客戶端向 ZooKeeper 服務器注冊 Watcher 的同時，會將 Watcher 對象存儲在客戶端的 WatchManager 中。ZooKeeper 服務器觸發(fā) Watcher 事件后，會向客戶端發(fā)送通知，客戶端線程從 WatchManager 中回調(diào) Watcher 執(zhí)行相應的功能。

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

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

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

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

        allZnodes = newZodeInfos;
    }

所以需要多帶帶運行一個進程，例如：

public class TestCuratorManager {

    public static void main(String[] args) {

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

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

總結(jié)

爬蟲框架 github 地址：github.com/fengzhizi71…

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

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