Javaベースの分散クローラーシステムの構築方法を段階的に説明します

[51CTO.com からのオリジナル記事] クローラー フレームワークを使用せずに、さまざまなソースから学び、MySQL、HBase などのさまざまな場所にデータを保存できる分散クローラー システムを実装しようとしました。

このシステムはインターフェース指向のコーディング思想に基づいて開発されているため、一定のスケーラビリティを備えています。興味のある友人は、コードを見るだけでその設計のアイデアを理解できます。

コードはまだ多くの場所で密接に結合されていますが、ある程度の時間と労力を費やせば、その大部分を抽出して構成することができます。

時間の制約により、JD.com と Suning.com の 2 つの Web サイトのクローラーのみを作成しました。ただし、さまざまな Web サイトに対してクローラーのランダム スケジュールを実装することは完全に可能です。コード構造から判断すると、Gome、Tmall などの商品クローラーの作成は難しくありませんが、ある程度の時間と労力がかかると予想されます。

なぜなら、Web ページ上のデータを解析するとき、たとえば Suning.com 製品の価格をクロールしていたとき、価格は非同期的に取得され、その API はデジタルの組み合わせの長い文字列だったからです。パターンを発見するのに数時間かかりました。もちろん、私には経験不足だったことを認めなければなりません。

基本的なデータ クロールに加えて、このシステムの設計では次の問題に重点を置いています。

• 配布を実現するにはどうすればいいですか?同じプログラムをパッケージ化して異なるノードに配布して実行しても、全体的なデータ クロールには影響しません。
• URL ランダム ループ スケジューリングを実装するにはどうすればよいでしょうか?核となるのは、異なる *** ドメイン名をランダム化することです。
• シード URL を URL リポジトリに定期的に追加するにはどうすればよいですか?クローラーシステムを停止させない目的を達成するため。
• クローラー ノード プログラムを監視し、電子メール アラートを送信する方法は?
• ランダム IP プロキシ ライブラリを実装するにはどうすればよいでしょうか?目的はポイント 2 と多少似ており、どちらもクローラー対策です。

以下ではこのシステムの基本について紹介します。コードには非常に詳細なコメントがあります。興味のある友人はコードを参照できます。最後に、クロール時のデータ分析をいくつか紹介します。

また、このクローラー システムは Java ベースで実装されていますが、言語自体が最も重要なわけではないことにも注意してください。興味のある方は、Python で実装してみてください。

分散クローラーシステムアーキテクチャ

システム全体のアーキテクチャは次のとおりです。

上記のアーキテクチャからわかるように、システム全体は主に 3 つの部分に分かれています。

• クローラーシステム
• URLディスパッチシステム
• 監視警報システム

クローラー システムは、データをクロールするために使用されます。システムは分散設計されているため、クローラー プログラム自体は異なるサーバー ノードで実行できます。

URL ディスパッチ システムの中核は URL ウェアハウスにあります。いわゆる URL ウェアハウスは、実際には Redis を使用してクロールする必要がある URL のリストを保存し、URL ディスパッチャーの特定の戦略に従ってその中の URL を消費します。この観点から見ると、URL リポジトリは実際には URL キューです。

監視アラームシステムは主にクローラーノードを監視します。並行して実行されるクローラー ノードの 1 つに障害が発生しても、全体的なデータ クロール自体には影響しません (クローラーの速度が低下するだけです)。ただし、ノード障害を受動的に検出するのではなく、能動的に通知を受信することを期待しています。

以下では、上記の 3 つの側面に焦点を当て、いくつかのコード スニペットを組み合わせて、システム全体の設計アイデアの基本的な紹介を行います。

クローラーシステム

クローラー システムは独立して実行されるプロセスです。クローラー システムを jar パッケージにパッケージ化し、実行のためにさまざまなノードに配布します。このように、データの並列クロールにより、クローラーの効率が向上します。 (注: ZooKeeper モニタリングはモニタリング アラーム システムに属し、URL ディスパッチャは URL ディスパッチ システムに属します)

ランダム IP プロキシ

ランダム IP プロキシを追加する主な目的は、アンチクローラーに対抗することです。したがって、IP プロキシ ライブラリがあり、http クライアントの構築時にさまざまなプロキシをランダムに使用できる場合、アンチクローラーとの戦いに非常に役立ちます。

システムで IP プロキシ ライブラリを使用するには、まず使用可能なプロキシ アドレス情報をテキスト ファイルに追加する必要があります。

 #IPプロキシリポジトリ.txt
 58.60.255.104:8118
 219.135.164.245:3128
 27.44.171.27:9999
 219.135.164.245:3128
 58.60.255.104:8118
 58.252.6.165:9000
 ......

上記のプロキシ IP は Xici Proxy から取得したプロキシ IP であり、利用できない可能性があることに注意してください。プロキシ IP をまとめて購入するためにお金を費やすことをお勧めします。これにより、プロキシ IP を見つけるための時間と労力を大幅に節約できます。

次に、http クライアントを構築するツール クラスで、ツール クラスが初めて使用されるときに、これらのプロキシ IP がメモリにロードされ、Java HashMap にロードされます。

 //IP アドレス プロキシ ライブラリ マップ
プライベート静的Map<String, Integer > IPProxyRepository = new HashMap<>();
プライベート静的String[] keysArray = null ; // keysArray はランダムなプロキシオブジェクトを生成するために使用されます
 
 /**
     * 初めて使用するときに、静的コードブロックを使用して IP プロキシライブラリをセットにロードします。
     */
静的{
    入力ストリーム= HttpUtil.class.getClassLoader().getResourceAsStream( "IPProxyRepository.txt" ); // プロキシIPを含むテキストをロードする
    //バッファストリームオブジェクトを構築する
    InputStreamReader isr = 新しい InputStreamReader( in );
    BufferedReader bfr = 新しい BufferedReader(isr);
    文字列行 = null ;
    試す {
        // 各行をループしてマップに追加します
        ((line = bfr.readLine()) != null ) の間 {
            文字列[]を分割 = line.split( ":" ); // 区切り文字として : を使用します。つまり、テキスト内のデータ形式は 192.168.1.1:4893 になります。
            文字列ホスト = split[0];
 intポート = Integer .valueOf(split[1]);
            IPProxyRepository.put(ホスト、ポート);
        }
 <String>を設定します。keys = IPProxyRepository.keySet();
        keysArray = keys.toArray(新しい文字列[ keys.size ()]); // keysArray はランダムなプロキシオブジェクトを生成するために使用されます
    } キャッチ (IOException e) {
        e.printStackTrace();
    } 
 
 }

その後、http クライアントを構築するたびに、まずマップをチェックしてプロキシ IP があるかどうかを確認します。はい、そうであればそれを使用してください。そうでない場合は、プロキシを使用しないでください。

 CloseableHttpClient httpClient = null ;
 HttpHost プロキシ = null ;
 if (IPProxyRepository.size () > 0) { // IPプロキシアドレスライブラリが空でない場合は、プロキシを設定します
    プロキシ = getRandomProxy();
    プロキシサーバに HTTP クライアントをインストールします。 // httpclient オブジェクトを作成する
}それ以外{
    カスタムビルド// httpclient オブジェクトを作成する
}
 HttpGet リクエスト = new HttpGet(url); // http getリクエストを構築
......

ランダム プロキシ オブジェクトは、次のメソッドを使用して生成されます。

 /**
     * プロキシオブジェクトをランダムに返す
     *
     * @戻る 
     */
公共 静的HttpHost getRandomProxy() {
    // ホスト:ポートをランダムに取得し、プロキシオブジェクトを構築します
    ランダム random = new Random();
    文字列ホスト = keysArray[random.nextInt(keysArray.length)];
 intポート = IPProxyRepository.get(ホスト);
    HttpHost プロキシ = 新しい HttpHost(ホスト、ポート); // httpプロキシを設定する
プロキシを返します。
 }

このように、上記の設計を通じて、ランダム IP プロキシの機能が基本的に実現されます。もちろん、改善できる点はまだたくさんあります。

たとえば、この IP プロキシを使用しているときにリクエストが失敗した場合、この状況を記録できますか?一定回数を超えるとプロキシライブラリから削除され、開発者や運用保守担当者が参照できるようにログが生成されます。これは完全に実行可能ですが、この手順は実行しません。

ウェブダウンローダー

Web ページ ダウンローダーは、Web ページからデータをダウンロードするために使用され、主に次のインターフェイスに基づいて開発されています。

 /**
 * Webデータのダウンロード
 */
パブリックインターフェースIDownload {
    /**
     * 指定されたURLのWebページデータをダウンロードします
     * @param URL
     * @戻る 
     */
パブリックページダウンロード(文字列URL);
 }

これに基づいて、システムには 1 つの http get ダウンローダーのみが実装されていますが、必要な機能も実行できます。

 /**
 * データダウンロード実装クラス
 */
パブリッククラスHttpGetDownloadImplはIDownloadを実装します{ 
 
    @オーバーライド
パブリックページダウンロード(文字列URL) {
        ページ page = new Page();
        文字列コンテンツ = HttpUtil.getHttpContent(url); // ウェブページのデータを取得する
        ページURLを設定します。
        ページのコンテンツを設定します。
ページに戻る;
    }
 }

ウェブページパーサー

Web ページ パーサーは、ダウンロードされた Web ページから目的のデータを解析し、それをオブジェクトに保存します。その後、データ ストレージ デバイスによってさらに処理され、さまざまな永続リポジトリに保存されます。以下のインターフェースに基づいて開発されています。

 /**
 * Webページデータ分析
 */
パブリックインターフェース IParser {
 public void parser(ページ page);
 }

Web ページ パーサーは、システム全体の開発において非常に重要なコンポーネントです。機能は複雑ではありませんが、多くのコードが含まれています。ショッピングモールや製品が異なれば、対応するパーサーも異なる場合があります。

したがって、JD.com が使用する Web ページ テンプレートは Suning.com が使用するものとは明らかに異なり、Tmall が使用する Web ページ テンプレートは JD.com が使用するものとは明らかに異なるため、特別なショッピング モール向けの製品を開発する必要があります。

したがって、これは完全にあなた自身の開発ニーズに基づいています。ただ、パーサー開発の過程で、重複するコードがいくつか見つかることがあります。この時点で、これらのコードを抽象化し、ツール クラスを開発できます。

現在、システムは JD.com と Suning.com の携帯電話製品データをクロールするため、次の 2 つの実装クラスが記述されています。

 /**
 * JD製品の実装クラスを分析する
 */
パブリッククラス JDHtmlParserImpl は IParser を実装します {
    ......
 } 
 
 /**
 * Suning.com ウェブサイト分析
 */
パブリッククラスSNHtmlParserImplはIParserを実装します{
    ......
 }

データストレージ

データ ストレージ デバイスは主に、Web ページ パーサーによって解析されたデータ オブジェクトをさまざまなテーブルに保存します。今回クロールした携帯電話製品のデータ オブジェクトは次の Page オブジェクトです。

 /**
 * ウェブページオブジェクト（主にウェブページのコンテンツと製品データを含む）
 */
パブリッククラス Page {
    プライベート文字列コンテンツ; //ウェブページコンテンツ
 
    プライベート文字列ID; // 製品ID
    プライベート文字列ソース; // 製品ソース
    プライベートStringブランド。 // 製品ブランド
    プライベート文字列タイトル; // 製品タイトル
    非公開浮動価格; // 製品価格
    プライベートintコメント数; // 製品コメントの数
    プライベート文字列 URL; // 製品アドレス
    プライベート文字列imgUrl; // 商品画像のURL
    プライベート文字列パラメータ; // 製品仕様パラメータ
 
    プライベートList<String> urls = new ArrayList<>(); // リストページを解析するときに解析された製品 URL を格納するコンテナ
}

同様に、MySQL では、テーブル データ構造は次のようになります。

 ----------------------------  
 -- 電話のテーブル構造 
 ----------------------------  
落とす テーブルが存在する場合、`phone`;
作成する テーブル`電話` (
  `id` varchar (30)文字  SETアームスシー8 NOT   NULLコメント'製品ID' 、
  `source` varchar (30) ではない  NULL COMMENT '製品ソース、例えば jd suning gome など。' 、
  `brand` varchar (30)デフォルト  NULLコメント「携帯電話ブランド」 、
  `title` varchar (255)デフォルト  NULL COMMENT '製品ページのモバイルタイトル' 、
  `price` float (10,2)デフォルト  NULL COMMENT '携帯電話の価格' 、
  `comment_count` varchar (30)デフォルト  NULL COMMENT '携帯電話のコメント' 、
  `url` varchar (500)デフォルト  NULL COMMENT '携帯電話の詳細住所' 、
  `img_url` varchar (500)デフォルト  NULL COMMENT '画像アドレス' 、
  `params` テキスト COMMENT '携帯電話パラメータ、json 形式で保存' 、
主要な キー(`id`,`source`)
 ) エンジン=InnoDBデフォルト文字セット=utf8;

HBase のテーブル構造は次のとおりです。

 ## cf1 ストア ID ソース 価格 コメント ブランド URL
 ## cf2はタイトルパラメータimgUrlを保存します
作成する  「電話」 、 「cf1」 、 「cf2」   
 
 ## HBaseシェルで作成されたテーブルを表示する
hbase(main):135:0>説明  '電話'  
テーブルフォンが有効です
電話
列ファミリーの説明
{ NAME => 'cf1' 、BLOOMFILTER => 'ROW' 、VERSIONS => '1' 、IN_MEMORY => 'false' 、KEEP_DELETED_CELLS => 'FALSE' 、DATA_BLOCK
 _ENCODING => 'NONE' 、TTL => 'FOREVER' 、COMPRESSION => 'NONE' 、MIN_VERSIONS => '0' 、BLOCKCACHE => 'true' 、BLOCKSIZE =>
 '65536' 、レプリケーションスコープ => '0' }
 { NAME => 'cf2' 、BLOOMFILTER => 'ROW' 、VERSIONS => '1' 、IN_MEMORY => 'false' 、KEEP_DELETED_CELLS => 'FALSE' 、DATA_BLOCK
 _ENCODING => 'NONE' 、TTL => 'FOREVER' 、COMPRESSION => 'NONE' 、MIN_VERSIONS => '0' 、BLOCKCACHE => 'true' 、BLOCKSIZE =>
 '65536' 、レプリケーションスコープ => '0' }
 0.0350秒で2行

つまり、HBase に cf1 と cf2 という 2 つの列ファミリが作成されます。 cf1 は、ID、ソース、価格、コメント、ブランド、URL フィールド情報を保存するために使用されます。 cf2 は、タイトル、パラメータ、imgUrl フィールド情報を保存するために使用されます。

異なるデータ ストアでは異なる実装クラスが使用されますが、それらはすべて同じインターフェイスに基づいて開発されます。

 /**
 * 製品データの保存
 */
パブリックインターフェース IStore {
パブリックvoid ストア (ページ page);
 }

そして、これを基にMySQLストレージ実装クラス、HBaseストレージ実装クラス、コンソール出力実装クラスを開発しました。たとえば、MySQL ストレージ実装クラスは、実際には単純なデータ挿入ステートメントです。

 /**
 * dbcデータベース接続プールを使用してmysqlテーブルにデータを書き込みます
 */
パブリッククラスMySQLStoreImplはIStoreを実装します{
    プライベート QueryRunner queryRunner = new QueryRunner(DBCPUtil.getDataSource()); 
 
    @オーバーライド
パブリックvoid store(ページ ページ) {
        文字列 sql = "insert into phone(id, source, brand, title, price, comment_count, url, img_url, params) values(?, ?, ?, ?, ?, ?, ?, ?, ?)" ;
        試す {
            queryRunner.update (sql、page.getId()、
                    ページ.getSource()、
                    page.getBrand(),
                    ページ.getTitle(),
                    page.getPrice(),
                    page.getCommentCount(),
                    ページ.getUrl()、
                    ページ.getImgUrl(),
                    ページ.getParams());
        } キャッチ (SQLException e) {
            e.printStackTrace();
        }
    }
 }

HBase ストレージ実装クラスは、HBase Java API のよく使用される挿入ステートメント コードです。

 ......
 // cf1:価格
Put 価格Put = 新しい Put(rowKey);
 // nullかどうかを確認する必要があります。そうでない場合は、null ポインタ例外が発生します。
 pricePut.addColumn(cf1, "price" .getBytes(), page.getPrice() != null ? String.valueOf(page.getPrice()).getBytes() : "" .getBytes());
 puts.add (pricePut);
 // cf1:コメント
コメントを入力します。Put = new Put(rowKey);
 commentPut.addColumn(cf1, "コメント" .getBytes(), page.getCommentCount() != null ? String.valueOf(page.getCommentCount()).getBytes() : "" .getBytes());
 puts.add (コメントPut);
 // cf1:ブランド
ブランドPut = new Put(rowKey);
 brandPut.addColumn(cf1, "brand" .getBytes(), page.getBrand() != null ? page.getBrand().getBytes() : "" .getBytes());
 puts.add (brandPut);
 ......

もちろん、データを保存する場所については、クローラーを初期化するときに手動で選択できます。

 // 3. メモリを挿入する
iSpider.setStore(新しい HBaseStoreImpl());

コードは、同時に複数の場所に保存できるようにはまだ書かれていません。現在のコード アーキテクチャによれば、これを実現するのは比較的簡単です。対応するコードを変更するだけです。

実際には、まずデータを MySQL に保存し、その後 Sqoop を介して HBase にインポートすることができます。詳しい操作については私が書いたSqoopの記事を参照してください。

データを HBase に保存する必要があることが確実な場合は、使用可能なクラスター環境があることを確認して、次の構成ドキュメントをクラスパスに追加することが重要です。

コアサイト.xml
 hbase サイト.xml
 hdfs-サイト.xml

ビッグデータに興味のある学生はぜひ試してみてください。これまで使用したことがない場合は、MySQL ストレージを使用してください。クローラー プログラムを初期化するときにのみ、MySQL ストレージを挿入する必要があります。

 // 3. メモリを挿入する
iSpider.setStore(新しい MySQLStoreImpl());

URLディスパッチシステム

URL スケジューリング システムは、クローラー システム全体の分散を実現するための架け橋であり、鍵となります。 URL スケジューリング システムを使用することで、クローラー システム全体がより効率的に (ストレージとして Redis を使用) URL をランダムに取得し、システム全体の分散を実現できます。

URL リポジトリ

アーキテクチャ図から、いわゆる URL ウェアハウスは Redis ウェアハウスに過ぎないことがわかります。つまり、システムでは URL アドレス リストを保存するために Redis が使用されています。

このようにして、プログラムが配布されることを保証できます。保存された URL が一意である限り、クローラー プログラムがいくつあっても、最終的に保存されるデータは一意であり、重複することはありません。

同時に、URL ウェアハウス内の URL アドレスを取得するための戦略はキューを通じて実装されます。これは、後で URL スケジューラの実装を通じて学習されます。

さらに、当社の URL リポジトリには、主に次のデータが保存されます。

シードURLリスト、Redisのデータ型はリストです

シード URL は永続的に保存されます。一定時間が経過すると、URL タイマーはシード URL を通じて URL を取得し、クローラーが使用する必要がある優先度の高い URL キューに挿入します。

これにより、クローラー プログラムの実行を終了せずに、継続的にデータをクロールできるようになります。

優先度の高いURLキュー、Redisデータ型が設定されている

高優先度 URL キューとは何ですか?実際には、リストの URL を保存するために使用されます。では、リスト URL とは何でしょうか?

簡単に言えば、リストには複数の製品が含まれます。 JD.com を例に、携帯電話リストを開きます。

このアドレスには、特定の製品の URL ではなく、クロールする必要がある複数のデータ (携帯電話製品) のリストが含まれています。

各高レベル URL を解析することで、多数の特定の製品 URL を取得できます。特定の製品 URL は低優先度 URL であり、低優先度 URL キューに保存されます。

このシステムを例にとると、保存されるデータは次のようになります。

 jd.com.higher
 --https://list.jd.com/list.html?cat=9987,653,655&page=1  
    ...
 suning.com.higher
 --https://list.suning.com/0-20006-0.html  
    ...

低優先度URLキュー、Redisデータ型が設定されている

優先度の低い URL は、実際には次の携帯電話製品のような特定の製品の URL です。

URL のデータをダウンロードして解析することで、必要なデータを取得できます。

このシステムを例にとると、保存されるデータは次のようになります。

 jd.com.lower  
 --https://item.jd.com/23545806622.html  
    ...
 suning.com.lower  
 --https://product.suning.com/0000000000/690128156.html  
    ...

URL ディスパッチャー

いわゆる URL ディスパッチャーは、URL ウェアハウス Java コードのディスパッチ戦略です。ただし、その核心はディスパッチにあるため、説明のために URL ディスパッチャ内に置かれます。現在、ディスパッチは次のインターフェースに基づいて開発されています。

 /**
 * URLリポジトリ
 * 主な機能:
 * 倉庫に URL を追加します (優先度の高いリスト、優先度の低い製品 URL)
 * 倉庫から URL を取得します (最初に優先度の高い URL を取得し、そうでない場合は優先度の低い URL を取得します)
 *
 */
パブリックインターフェースIRepository { 
 
    /**
     * URLの取得方法
     * 倉庫から URL を取得します (最初に優先度の高い URL を取得し、そうでない場合は優先度の低い URL を取得します)
     * @戻る 
     */
パブリック文字列 poll(); 
 
    /**
     * 製品リストの URL を高優先度リストに追加します
     * @param 高URL
     */
パブリックvoid offerHigher(String highUrl); 
 
    /**
     * 低優先度リストに商品URLを追加する
     * @param 低URL
     */
パブリックvoid offerLower(String lowUrl); 
 
 }

URL リポジトリとして Redis をベースにした実装は次のとおりです。

 /**
 * Redis をベースにしたフルネットワーク クローラー。クローラー URL をランダムに取得します。
 *
 * Redis に URL を保存するために使用されるデータ構造は次のとおりです。
 * 1. クロールする必要があるドメイン名セット（ストレージデータ型が設定されており、最初にRedisに追加する必要があります）
 *鍵 
 * スパイダー.ウェブサイト.ドメイン
 * 値(設定)
 * jd.com suning.com gome.com
 *キーは定数オブジェクト SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY から取得されます
 * 2. 各ドメイン名に対応する高優先度および低優先度の URL キュー (保存データ タイプはリストであり、クローラー プログラムがシード URL を解析した後に動的に追加されます)
 *鍵 
 * jd.com.higher
 * jd.com.lower  
 * suning.com.higher
 * suning.com.lower  
 * gome.com.higher
 *下へ 
 * 値(リスト)
 * 解析する必要がある対応するURLリスト
 *キーは、定数 SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX または SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX にランダムなドメイン名を追加することによって取得されます。
 * 3. シードURLのリスト
 *鍵 
 * スパイダーシードのURL
 * 値(リスト)
 * クロールするデータのシードURL
 *キーは定数SpiderConstants.SPIDER_SEED_URLS_KEYから取得されます
 *
 * シード URL リスト内の URL は、URL スケジューラによって高優先度 URL キューと低優先度 URL キューに定期的に追加されます。
 */
パブリッククラス RandomRedisRepositoryImpl は IRepository を実装します { 
 
    /**
     * 施工方法
     */
パブリックRandomRedisRepositoryImpl() {
        初期化();
    } 
 
    /**
     * 初期化方法、初期化時に、まずRedis内の高優先度と低優先度のURLキューをすべて削除します
     * そうでない場合、最後の URL キュー内の URL が消費されず、停止して次の実行を開始すると、URL ウェアハウスに重複した URL が存在することになります。
     */
パブリックvoid init() {
        ジェディス jedis = JedisUtil.getJedis();
 <String> domains = jedis.smembers(SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY);を設定します。
        文字列higherUrlKey;
        文字列lowerUrlKey;
 for (文字列ドメイン:ドメイン) {
            より高いUrlKey = ドメイン + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX;
            lowerUrlKey = ドメイン + SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX;
            jedis.del(higherUrlKey, lowerUrlKey);
        }
        JedisUtil.returnJedis(jedis);
    } 
 
    /**
     * キューから URL を取得します。現在の戦略は次のとおりです。
     * 1. 優先度の高いURLキューから最初に取得する
     * 2. 優先度の低いURLキューから取得する
     * 実際のシナリオでは、まずリストのURLを解析し、次に製品のURLを解析する必要があります。
     * ただし、分散マルチスレッド環境では、優先度の高いURLキューのどこかの時点で、これが完全に保証されるわけではないことに注意してください。
     * URL は消費されていますが、プログラムはまだ次の優先度の高い URL を解析中です。このとき、他のスレッドは高優先度キューの URL を確実に取得できなくなります。
     ※このとき、優先度の低いキューにあるURLが取得されます。これは分析を検討するときに特に重要です。
     * @戻る 
     */
    @オーバーライド
パブリック文字列ポーリング（）{
        //セットからランダムに***ドメイン名を取得します
        ジェディス jedis = JedisUtil.getJedis();
        文字列 randomDomain = jedis.srandmember(SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY); // jd.com
        文字列キー= randomDomain + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX; // jd.com.higher
        文字列 url = jedis.lpop(キー);
        if(url == null ) { // nullの場合は低優先度から取得する
キー= randomDomain + SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX; // jd.com より。より低い 
            url = jedis.lpop(キー);
        }
        JedisUtil.returnJedis(jedis);
 URLを返します。
    } 
 
    /**
     * URLを高優先度URLキューに追加します
     * @param 高URL
     */
    @オーバーライド
パブリックvoid offerHigher(String highUrl) {
        オファーUrl(highUrl、SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX);
    } 
 
    /**
     * 低優先度のURLキューにURLを追加する
     * @param 低URL
     */
    @オーバーライド
パブリックvoid offerLower(String lowUrl) {
        低いUrlを提供します。
    } 
 
    /**
     * offerHigher と offerLower から抽象化された URL を追加するための汎用メソッド
     * @param url 追加するURL
     * @param urlTypeSuffix URL タイプのサフィックス .higher または.lower  
     */
パブリックvoid offerUrl(文字列 url、文字列 urlTypeSuffix) {
        ジェディス jedis = JedisUtil.getJedis();
        文字列ドメイン = SpiderUtil.getTopDomain(url); // URL に対応するトップドメイン名を取得します (例: jd.com)
        文字列キー= ドメイン + urlTypeSuffix; // jd.com.higher などのURL キューのキーを連結します。
        jedis.lpush(キー、 URL ); // URL を URL キューに追加します
        JedisUtil.returnJedis(jedis);
    }
 }

コード分​​析を通じて、核心は URL ウェアハウス (Redis) で URL をスケジュールする方法にあることもわかります。

URL タイマー

一定時間が経過すると、優先度の高い URL キューと優先度の低い URL キューの両方の URL が消費されます。

人間の介入を減らしながらプログラムがデータのクロールを継続できるようにするには、事前に Redis にシード URL を挿入し、URL タイマーがシード URL から URL を取り出して、定期的に高優先度 URL キューに格納するようにします。これにより、プログラムが一定間隔で中断することなくデータをクロールするという目的を達成できます。

URL が使用された後、データを継続的にクロールする必要があるかどうかは、個人のビジネス ニーズによって異なります。したがって、この手順は必須ではありませんが、このような操作も提供されます。

実際のところ、クロールする必要があるデータは定期的に更新されるからです。クロールするデータを定期的に更新したい場合、タイマーが非常に重要な役割を果たします。

ただし、ループ内でデータを繰り返しクロールする必要があると判断された場合は、メモリ実装の設計時に重複データの問題を考慮する必要があり、つまり重複データを更新する必要があることに注意してください。

私が設計したストレージデバイスには、現時点ではこの機能は含まれていません。興味のある友人は自分でそれを実装することができます。データを挿入する前に、データベースにデータが存在するかどうかを確認するだけで済みます。

注意すべきもう 1 つの点は、URL タイマーは独立したプロセスであり、個別に開始する必要があることです。

タイマーはQuartzに基づいて実装されています。そのジョブのコードは次のとおりです。

 /**
 * シードURLをURLウェアハウスから毎日定期的に取得し、優先度の高いリストに追加します
 */
パブリッククラスUrlJobはJobを実装します{ 
 
    // log4j ログ
    プライベート Logger ロガー = LoggerFactory.getLogger(UrlJob.class); 
 
    @オーバーライド
パブリックvoid実行(JobExecutionContext コンテキスト) は JobExecutionException をスローします {
        /**
         * 1. 指定されたURLシードリポジトリからシードURLを取得します
         * 2. シードURLを高優先度リストに追加する
         */
        ジェディス jedis = JedisUtil.getJedis();
 <String> seedUrls = jedis.smembers(SpiderConstants.SPIDER_SEED_URLS_KEY);を設定します。 // spider.seed.urls Redis データ型は、シード URL の繰り返し追加を防ぐために設定されています
(文字列 seedUrl : seedUrls) {
            文字列ドメイン = SpiderUtil.getTopDomain(seedUrl); // *** シード URL のドメイン名
            jedis.sadd(ドメイン + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX、seedUrl);
            logger.info( "シードを取得:{}" , seedUrl);
        }
        JedisUtil.returnJedis(jedis);
 // システム。 out .println( "スケジューラジョブテスト..." );
    } 
 
 }

スケジューラの実装は次のとおりです。

 /**
 * URLタイミングスケジューラ、シードURLをURL対応ウェアハウスに定期的に保存
 *
 * 業務規定：毎日午前1時10分にシードURLを倉庫に保存する
 */
パブリッククラス UrlJobScheduler { 
 
パブリックUrlJobScheduler() {
        初期化();
    } 
 
    /**
     * スケジューラを初期化する
     */
パブリックvoid init() {
        試す {
            スケジューラ scheduler = StdSchedulerFactory.getDefaultScheduler(); 
 
            // 次の開始メソッドが実行されない場合、タスクのスケジュールは開始されません
            スケジューラを開始します。 
 
            文字列名= "URL_SCHEDULER_JOB" ;
            文字列グループ= "URL_SCHEDULER_JOB_GROUP" ;
            JobDetail jobDetail = new JobDetail(名前、グループ、 UrlJob.class);
            文字列 cronExpression = "0 10 1 * * ?" ;
トリガー トリガー= 新しい CronTrigger(名前、グループ、cronExpression); 
 
            //タスクをスケジュールする
            スケジューラ.scheduleJob(ジョブの詳細、トリガー); 
 
        } キャッチ (SchedulerException e) {
            e.printStackTrace();
        } キャッチ (ParseException e) {
            e.printStackTrace();
        }
    } 
 
公共 静的void main(String[] args) {
        URLJobScheduler urlJobScheduler = new UrlJobScheduler();
        urlJobScheduler.start();
    } 
 
    /**
     * スケジュールされたタスク
     * 指定された倉庫からシードURLを毎日定期的に取得し、優先度の高いURLリストに保存する必要があるため
     * これは中断されないプロセスなので、止めることはできません
     */
    プライベートvoid start() {
        （真）の間{ 
 
        }
    }
 }

監視警報システム

監視アラーム システムの追加は、実際にはクローラー プログラムが継続的に実行されている可能性があるため、ユーザーがノード ダウンタイムを受動的に検出するのではなく、積極的に検出できるようにすることが主な目的です。

また、クローラー プログラムを複数のノードに展開するため、ノードを監視し、問題が発生したときにそれを適時に検出して修正する必要があります。監視アラーム システムは独立したプロセスであり、別途開始する必要があることに注意してください。

根拠

まず、ZooKeeper に /ispider ノードを作成する必要があります。

 [zk: localhost:2181(接続済み) 1] /ispider を作成しますispider
 /ispider を作成しました

監視およびアラーム システムの開発は、主に ZooKeeper の実装に依存しています。監視プログラムは、ZooKeeper の下のノード ディレクトリを監視します。

 [zk: localhost:2181(接続済み) 0] ls /ispider  
 []

クローラー プログラムが起動すると、このノード ディレクトリの下に一時ノード ディレクトリが登録されます。

 [zk: localhost:2181(接続済み) 0] ls /ispider  
 [192.168.43.166]

ノードがクラッシュすると、一時ノード ディレクトリは ZooKeeper によって削除されます。

 [zk: localhost:2181(接続済み) 0] ls /ispider 
 
 []

同時に、ノード ディレクトリ /ispider を監視しているため、ZooKeeper がその下のノード ディレクトリを削除すると (またはノード ディレクトリを追加すると)、ZooKeeper は監視プログラムに通知を送信します。

つまり、監視プログラムはコールバックを取得し、コールバック プログラムでアラーム システム アクションを実行して、監視アラーム機能を完了します。

ZooKeeper Java API 使用上の注意

ZooKeeper のネイティブ Java API を使用できます。私が作成した別の RPC フレームワークのネイティブ API を使用しました (基礎レイヤーはリモート通信を実現するために Netty に基づいています)。

ただし、コードは明らかにはるかに複雑になるため、使いやすくするために ZooKeeper についてさらに学ぶ必要があります。

そのため、開発の難易度を軽減するために、ここではサードパーティのカプセル化された API、つまり curator を使用して ZooKeeper クライアント プログラムを開発します。

クローラーシステムZooKeeper登録

クローラー システムを起動すると、プログラムは ZooKeeper クライアントを起動し、独自のノード情報 (主に IP アドレス) を ZooKeeper に登録します。

また、/ispider ノード ディレクトリに、クローラー プログラムが配置されているノードの IP アドレスにちなんで名付けられたノード (/ispider/192.168.43.116 など) を作成します。実装コードは次のとおりです。

 /**
 * 登録 zk
 */
プライベートvoidレジスタZK() {
    文字列 zkStr = "uplooking01:2181、uplooking02:2181、uplooking03:2181" ;
 intベーススリープ時間Ms = 1000;
最大再試行回数 = 3;
    再試行ポリシー retryPolicy = 新しい ExponentialBackoffRetry(baseSleepTimeMs、maxRetries);
    CuratorFramework キュレーター = CuratorFrameworkFactory.newClient(zkStr, retryPolicy);
    キュレーターを起動します。
    文字列 ip = null ;
    試す {
        // zkの特定のディレクトリに登録する ノードを書き込む ノードを作成する
        ip = InetAddress.getLocalHost().getHostAddress();
        curator.create ().withMode(CreateMode.EPHEMERAL).forPath( "/ispider/" + ip, ip.getBytes());
    } キャッチ (UnknownHostException e) {
        e.printStackTrace();
    } キャッチ (例外 e) {
        e.printStackTrace();
    }
 }

作成したノードは一時的なノードであることに注意してください。監視・警報機能を実現するには、一時ノードにする必要があります。

モニタリングプログラム

まず、ZooKeeper でノード ディレクトリを監視する必要があります。私たちのシステムでは、ノード ディレクトリ /ispider を監視するように設計されています。

パブリックスパイダーモニタータスク() {
    文字列 zkStr = "uplooking01:2181、uplooking02:2181、uplooking03:2181" ;
 intベーススリープ時間Ms = 1000;
最大再試行回数 = 3;
    再試行ポリシー retryPolicy = 新しい ExponentialBackoffRetry(baseSleepTimeMs、maxRetries);
    curator = CuratorFrameworkFactory.newClient(zkStr, retryPolicy);
    キュレーターを起動します。
    試す {
        previousNodes = curator.getChildren().usingWatcher(this).forPath( "/ispider" );
    } キャッチ (例外 e) {
        e.printStackTrace();
    }
 }

上記では、通知を受信するためのコールバック プログラムである ZooKeeper のウォッチャーを登録しています。このプログラムでは、アラーム ロジックが実行されます。

 /**
 * このメソッドは、監視対象のzkに対応するディレクトリが変更されたときに呼び出されます。
 * 現在の***ノードステータスを取得し、***ノードステータスを初期または以前のノードステータスと比較すると、ノード変更の原因がわかります。
 * @param イベント
 */
 @オーバーライド
パブリックvoid プロセス(WatchedEvent イベント) {
    試す {
        リスト<String> currentNodes = curator.getChildren().usingWatcher(this).forPath( "/ispider" );
        // HashSet<String> previousNodesSet = new HashSet<>(previousNodes);
        if( currentNodes.size () > previousNodes.size ()) { // ***のノードサービスが以前のノードサービスの数を超え、新しいノードが追加されます
(文字列ノード: currentNodes) {
                if(!previousNodes. contains (ノード)) {
                    // 現在のノードは新しく追加されたノードです
                    logger.info( "----新しいクローラーノード {} が追加されました" , node);
                }
            }
        } else if(currentNodes.size ( ) < previousNodes.size ()) { //ノードがダウンしています。アラートメールまたは SMS を送信します。
 (文字列ノード: previousNodes) {
                if(!currentNodes. contains (ノード)) {
                    // 現在のノードがダウンしているため、メールを送信する必要があります
                    logger.info( "----クローラーノード {} がクラッシュしました" , node);
                    MailUtil.sendMail( "クローラーノードがダウンしています。クローラーノードのステータスを手動で確認してください。ノード情報は次のとおりです: " , node);
                }
            }
        } // ドロップされたアイテムと新しく追加されたアイテムの数は完全に同じです。この状況は上記に含まれていません。興味のある友達は、この特別な状況を含む監視を直接実装できます
        以前のnodes = currentNodes; //前のノードリストを更新して最新のノードリストになる
    } キャッチ (例外 e) {
        e.printStackTrace();
    }
    //ネイティブAPIは再度監視する必要があります。各監視は1回しか有効になるためです。
    //しかし、これはキュレーターAPIを使用するときは必要ありません
}

もちろん、ノードがダウンしているかどうかを判断するための上記のロジックにはまだいくつかの問題があります。上記のロジックによれば、新しいノードを追加してノードを削除するイベントが同時に発生する場合、判断することはできません。したがって、より精度が必要な場合は、上記のプログラムコードを変更できます。

メール送信モジュール

テンプレートコードを使用できますが、使用する場合は、独自のメールアドレスを送信者として使用してください。

以下は、クローラーノードがハングアップしたときに受け取ったメールです。

実際、SMSサービスを購入すると、SMS APIを介して携帯電話にテキストメッセージを送信することもできます。

練習：jd.comとSuning.comのモバイル製品データをクロールします

このシステムを導入したときに述べたように、私はJD.comとSuning.comのWebページパーサーのみを書いたので、次のステップはネットワーク全体の携帯電話製品データをクロールすることです。

環境説明

RedisおよびZookeeperサービスが利用できるようにする必要があります。さらに、HBaseを使用してデータを保存する必要がある場合は、Hadoopクラスター内のHBaseが利用可能であり、関連する構成ファイルがCrawlerプログラムのClassPathに追加されていることを確認する必要があります。

注意すべきもう1つのことは、URLタイマーと監視アラームシステムが個別のプロセスとして実行され、オプションでもあることです。

クローラーの結果

データを2回クロールし、それをそれぞれMySQLとHBaseに保存しようとしましたが、次のデータが与えられました。

mysqlに保存します

mysql> select  電話から（*） 。
 + ----------+  
 | count （*）|
 + ----------+  
 | 12052 |
 + ----------+  
 1列 セット  
 
 mysql> select   count （*） source = ' jd.com ' ;
 + ----------+  
 | count （*）|
 + ----------+  
 | 9578 |
 + ----------+  
 1列 セット  
 
 mysql> select   count （*） source = 'サンニング
.com ';
 + ----------+  
 | count （*）|
 + ----------+  
 | 2474 |
 + ----------+  
 1列 セット 

視覚化ツールでデータを表示します。

hbaseに保存します

hbase（メイン）：225：0* count   '電話'  
現在 カウント：1000、行：11155386088_jd.com
現在 カウント：2000、行：136191393_Suning.com
現在 カウント：3000、行：16893837301_jd.com
現在 カウント：4000、行：19036619855_jd.com
現在 カウント：5000、行：1983786945_jd.com
現在 カウント：6000、行：1997392141_jd.com
現在 カウント：7000、行：21798495372_jd.com
現在 カウント：8000、行：24154264902_jd.com
現在 カウント：9000、行：25687565618_jd.com
現在 カウント：10000、行：26458674797_jd.com
現在 カウント：11000、行：617169906_Suning.com
現在 カウント：12000、行：769705049_Suning.com
 1.5720秒で12348行
=> 12348

HDFSでデータを表示します。

データ量と実際の状況の分析

JD：JD携帯電話のリストは約160ページで、各リストには60の製品データがあるため、合計金額は約9,600で、データは基本的にそれに沿っています。

後で、ログ分析を通じて、失われたデータは一般に接続タイムアウトによって引き起こされることがわかります。したがって、クローラー環境を選択するときは、優れたネットワーク環境を持つホストでそれを行うことをお勧めします。

同時に、IPプロキシアドレスライブラリがある場合は、より良いでしょう。さらに、接続タイムアウトでは、プログラムでさらに制御できます。

データのクロールに失敗したURLが発生したら、それを再試行のURLキューに追加できます。現在、この機能を行っていません。興味のある学生はそれを試すことができます。

Suning.com：Suningのデータを見てみましょう。約100ページの携帯電話リストがあり、各ページには60個の製品データもあるため、合計金額は約6,000です。

しかし、私たちのデータは3000の順序にすぎないことがわかります（欠落しているものは、頻繁にrawうによって引き起こされる接続障害の問題です）。これはなぜでしょうか?

これは、Suningのリストページを開いた後、最初に30の製品をロードするためです。マウスがスライドすると、他の30の製品データが別のAPIを介してロードされます。これは、各リストページに当てはまります。したがって、実際、製品データの半分が欠落しており、rawいされていません。

この理由を知った後、達成することは難しくありませんが、時間の制約のために、私はもうそれをしませんでした。興味のある友達はそれをすることができます。

ログによるクローラーシステムのパフォーマンスの分析

クローラーシステムでは、Webページのダウンロード、データ分析などのすべての重要な場所が記録されるため、関連する時間パラメーターをログを通して大まかに分析できます。

 2018-04-01 21:26:03 [pool-1-thread-1] [cn.xpleaf.spider.utils.httputil] [情報]  -  Webページをダウンロード：https：//list.jd.com/list.html？cat = 9987,653,6555r： 590 ms： 590 ms 、 proxption  
 2018-04-01 21:26:03 [Pool-1-Thread-1] [cn.xpleaf.spider.core.parser.impl.jdhtmlparserimpl] [情報]  - パーサーリストページ：https://list.jd.com/list.html？
 2018-04-01 21:26:03 [Pool-1-Thread-3] [cn.xpleaf.spider.core.parser.impl.snhtmlparserimpl] [情報]  - パーサーリストページ：https://list.suning.com/0-20006-0.html
 2018-04-01 21:26:04 [Pool-1-Thread-5] [cn.xpleaf.spider.utils.httputil] [情報]  -  Webページをダウンロード：https：//item.jd.com/6737464.html、Consume：219 MS、Null：Null：Null：Null：Null：Null：Null：Null：Null：Null：Null： Null ： Null： Null  
 2018-04-01 21:26:04 [Pool-1-Thread-2] [cn.xpleaf.spider.utils.httputil] [情報]  -  Webページをダウンロード：https：//list.jd.com/list.html？cat = 9987,653,655＆page=2＆sort-sort = sort_asct_asct_asct_asct_asct_asp_asc 276ミリ秒、プロキシ情報： null ： null  
 2018-04-01 21:26:04 [Pool-1-Thread-4] [cn.xpleaf.spider.utils.httputil] [情報]  -  Webページをダウンロード：https：//list.suning.com/0-20006-99.html、Consume：300 MS、Null：Null ：Null：Null：Null ：Null ： Null  
 2018-04-01 21:26:04 [Pool-1-Thread-4] [cn.xpleaf.spider.core.parser.impl.snhtmlparserimpl] [情報]  - パーサーリストページ：https://list.suning.com/0-20006-99.html
 ......
 2018-04-01 21:27:49 [Pool-1-Thread-3] [cn.xpleaf.spider.utils.httputil] [情報]  -  Webページをダウンロード：https://club.jd.com/comment/ProductCommentsummaries。 Action ？ReferenceIDS = 23934388891、消費時間：176 ms、プロキシ情報： null ： null  
 2018-04-01 21:27:49 [pool-1-thread-3] [cn.xpleaf.spider.core.parser.Impl.JDHtmlParserImpl] [INFO] - 解析商品页面:https://item.jd.com/23934388891.html, 消耗时长:413ms
 2018-04-01 21:27:49 [Pool-1-Thread-2] [cn.xpleaf.spider.utils.httputil] [情報]  -  https：//review.suning.com/ajax/review_satisfy/general-000000000000000100179333333333333079092 、プロキシ情報：null：null  
 2018-04-01 21:27:49 [Pool-1-Thread-2] [cn.xpleaf.spider.core.parser.impl.snhtmlparserimpl] [情報]  - 分析製品ページ：https://product.suning.com/0070079092/10017779337.htmp
 ......

平均して、製品Webページのデータをダウンロードする時間は200〜500ミリ秒の範囲です。もちろん、これは当時のネットワークの状況に依存します。

さらに、製品をクロールするための時間データを実際に計算する場合は、ログの下のデータで計算できます。

• 製品ページのデータをダウンロードする時間
• 価格データを取得する時間
• コメントデータを取得する時間

私のホスト（CPU：E5 10コア、メモリ：32GB、仮想マシン、3つの仮想マシンがそれぞれ有効になっています）では、状況は次のとおりです。

3つのノードを使用すると、それに応じて時間が1/3に縮小しないことがわかります。これは、この時点でクローラーのパフォーマンスに影響を与える主な問題は、多数のノード、多数のスレッド、多数のネットワーク要求があるネットワークの問題であるためです。

ただし、帯域幅は確実であり、プロキシが使用されない場合、頻繁にリクエストが増加し、接続の障害も増加し、時間に特定の影響を与えます。ランダムプロキシライブラリを使用すると、状況がはるかに良くなります。

しかし、クローラーノードを水平スケールで追加した後、クローラーの時間を大幅に短縮できることは確かです。これは、分散クローラーシステムの利点でもあります。

Crawler Systemsで使用される反anti-Crawler戦略

クローラーシステム全体の設計では、主に反クローラーの目的を達成するために次の戦略が使用されています。

• プロキシを使用してアクセス - > IPプロキシライブラリ、ランダムIPプロキシ。
• ランダム***ドメインURLアクセス - > URLスケジューリングシステム。
• 各スレッドは、1つの製品データの睡眠をクロールする前に短時間睡眠をcrawっています。

要約する

このシステムはJavaに基づいて実装されていることに注意する必要がありますが、個人的には言語自体がまだ問題ではなく、コアはシステム全体の設計と理解にあると思います。

私はこの記事を書き、そのような分散クローラーシステムのアーキテクチャを全員と共有しました。ソースコードに興味がある場合は、GitHubで確認できます。

[[228686]]

Ye Yonghao、ビッグデータエンジニア、Huawei Hcie-RS認定エンジニア。彼はHuaweiやNeteaseなどの企業で働いており、現在、ビッグデータの分野での学習と研究に焦点を当てています。

[51CTO オリジナル記事、パートナーサイトに転載する場合は、元の著者とソースを 51CTO.com として明記してください]