Java マルチスレッドクローラーと分散クローラーアーキテクチャの調査

これは Java クローラーシリーズの 5 番目です。前回の記事では、Java クローラーサーバーがブロックされましたが、慌てずにサーバーを変更しましょう。主に IP ブロッキングとそれに対応する方法に焦点を当てて、クローラー対策戦略とクローラー対策方法について簡単に説明しました。これまでのいくつかの記事では、クローラーに関連するほぼすべての基本的な知識をカバーしました。この記事では、クローラーのアーキテクチャについて説明します。

[[279402]]

前の章では、クローラープログラムはすべてシングルスレッドでした。クローラープログラムをデバッグすると、シングルスレッドクローラーでは問題は発生しません。ただし、シングルスレッドクローラーを使用してオンライン環境で Web ページを収集する場合、シングルスレッドによって 2 つの致命的な問題が発生します。

収集効率は非常に低く、各スレッドはシリアルです。次の実行アクションは、前のアクションが完了するまで待機してから実行する必要があります。
サーバーのCPUの使用率は高くありません。考えてみてください。当社のサーバーはすべて 8 コア、16G、32G を搭載しています。スレッドを 1 つだけ実行するのは無駄ではないでしょうか?

オンライン環境は、結果が正しく抽出できる限り、収集効率を気にしないローカルテストのようなものであってはなりません。時は金なりと言われるこの時代では、ゆっくりとデータを収集する時間を与えることは不可能なので、シングルスレッドのクローラープログラムは機能しません。収集効率を改善し、コンピュータの使用率を高めるには、シングルスレッドモードをマルチスレッドモードに変更する必要があります。

マルチスレッドクローラープログラムの設計は、シングルスレッドクローラープログラムの設計よりもはるかに複雑です。ただし、高い同時実行性の下でデータセキュリティを確保する必要のある他のビジネスとは異なり、マルチスレッドクローラーでは各ページが独立したエンティティと見なすことができるため、データセキュリティに対する要件はそれほど高くありません。マルチスレッドクローラーをうまく機能させるには、2 つのことを適切に行う必要があります。1 つ目は、収集する URL の統一されたメンテナンスであり、2 つ目は URL の重複排除です。これら2点について簡単にお話ししましょう。

収集するURLを維持する

マルチスレッドクローラープログラムは、各スレッドが収集する独自の URL を維持するシングルスレッドプログラムのようにはなりません。この場合、各スレッドによって収集される Web ページは同じになります。これはマルチスレッドのコレクションではありませんが、ページを複数回収集しています。このため、収集する URL を統一的に管理する必要があります。各スレッドは、統合 URL メンテナンスから収集された URL を受け取り、収集タスクを完了します。ページ上で新しい URL リンクが見つかった場合、そのリンクは統合 URL によって管理されるコンテナーに追加されます。統合 URL メンテナンスに適したコンテナーをいくつか次に示します。

LinkedBlockingQueueなどのJDKのセキュアキュー
RedisやMongodbなどの高性能NoSQL
MQ メッセージミドルウェア

URL重複排除

URL 重複排除も、マルチスレッド収集における重要なステップです。重複を排除しないと、重複した URL が大量に収集され、収集効率が向上しません。たとえば、ページ分けされたニュースリストの場合、最初のページを収集するとページ 2、3、4、5 へのリンクが取得され、2 番目のページを収集するとページ 1、3、4、5 へのリンクが取得されます。収集する URL キューにはリストページリンクが大量に存在し、収集が繰り返され、無限ループに陥る可能性があるため、URL の重複排除が必要になります。 URL の重複を排除する方法は多数あります。よく使用される方法をいくつか紹介します。

Redis、MongoDBなどの重複排除用データベースにURLを保存する
ハッシュセットなどのハッシュテーブルにURLを入れて重複を削除します。
重複を削除するには、MD5 の後に URL をハッシュテーブルに保存します。上記の方法と比較して、スペースを節約できます
重複を削除するには、Bloom Filter を使用します。この方法は多くのスペースを節約できますが、それほど正確ではありません。

マルチスレッドクローラーに関する 2 つの重要な知識ポイントは誰もが知っています。以下に、シンプルなマルチスレッドクローラーアーキテクチャ図を示します。

マルチスレッドクローラーアーキテクチャ図

上記では、主にマルチスレッドクローラーのアーキテクチャ設計について学習しました。次に、Java マルチスレッドクローラーを試してみましょう。 Java マルチスレッドクローラーを練習するために、Hupu News のコレクションを例に挙げます。 Java マルチスレッドクローラーは、収集する URL を維持し、重複する URL を削除するように設計されています。ここではデモンストレーションのみなので、JDK の組み込みコンテナーを使用して完了させます。収集する URL を維持するためのコンテナとして LinkedBlockingQueue を使用し、重複する URL を削除するためのコンテナとして HashSet を使用します。以下は、Java マルチスレッドクローラーのコアコードです。詳細なコードは GitHub にアップロードされており、そのアドレスは記事の最後にあります。

 /**
 * マルチスレッドクローラー
 */
パブリッククラスThreadCrawlerはRunnableを実装します{
    // 収集された記事の数
    プライベート最終AtomicLong pageCount = new AtomicLong(0);
    // リストページリンクの正規表現
公共 静的最終文字列 URL_LIST = "https://voice.hupu.com/nba" ;
    保護されたロガー logger = LoggerFactory.getLogger(getClass());
    // 収集するキュー
    LinkedBlockingQueue<文字列> タスクキュー;
    // 収集されたリンクのリスト
    HashSet<String> が訪問されました。
    //スレッドプール
    カウント可能なスレッドプール スレッドプール;
    /**
     *
     * @param url 開始ページ
     * @param threadNum スレッド数
     * @throws 中断例外
     */
パブリックThreadCrawler(String url, int threadNum)はInterruptedExceptionをスローします{
        this.taskQueue = 新しい LinkedBlockingQueue<>();
        this.threadPool = 新しい CountableThreadPool(threadNum);
        this.visited = 新しい HashSet<>();
        // 収集するキューに開始ページを追加します
        タスクキューにURLを書き込みます。
    } 
 
    @オーバーライド
パブリックボイド実行（）{
        logger.info( "スパイダーが起動しました!" );
        while (!Thread.currentThread().isInterrupted()) {
            // キューから収集するURLを取得します
            最終的な文字列リクエスト = taskQueue.poll();
            // リクエストが空で、現在のスレッドに実行中のスレッドがない場合
            リクエストがnullの場合
                （threadPool.getThreadAlive() == 0）の場合{
                    壊す;
                }
            }それ以外{
                //コレクションタスクを実行する
                スレッドプール.execute (新しいRunnable() {
                    @オーバーライド
パブリックボイド実行（）{
                        試す {
                            リクエストを処理します。
                        } キャッチ (例外 e) {
                            logger.error( "プロセスリクエスト " + リクエスト + "エラー" , e);
                        ついに
                            // コレクションページ +1
                            ページカウントを増加して取得します。
                        }
                    }
                });
            }
        }
        スレッドプールをシャットダウンします。
        logger.info( "スパイダーが閉じました! {} ページがダウンロードされました。" , pageCount.get());
    } 
 
    /**
     * 収集リクエストの処理
     * @param URL
     */
    保護されたvoid processRequest(文字列url) {
        // リストページかどうかを判定する
        url.matches(URL_LIST) の場合
            // リストページから詳細ページリンクを解析し、収集するURLキューに追加します
            タスクキューを処理します(url);
        }それ以外{
            // Webページを解析する
            processPage(url);
        }
    }
    /**
     * リンク集の処理
     * リストページを処理し、URLをキューに追加します
     *
     * @param URL
     */
    保護されたvoid processTaskQueue(文字列url) {
        試す {
            ドキュメント doc = Jsoup.connect (url).get();
            //詳細ページリンク
            要素 elements = doc。 ( " div.news-list > ul > li > div.list-hd > h4 > a "を選択);
            elements.stream().forEach((要素 -> {
                文字列リクエスト = element.attr( "href" );
                // リンクがキューまたは収集されたセット内に存在するかどうかを確認します。そうでない場合は、キューに追加します。
                if (!visited. contains (request) && !taskQueue. contains (request)) {
                    試す {
                        タスクキューにリクエストを書き込みます。
                    } キャッチ (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }));
            // リストページリンク
            要素 list_urls = doc.select ( "div.voice-paging > a" );
            list_urls.stream().forEach((要素 -> {
                文字列リクエスト = element.absUrl( "href" );
                // 抽出するリストリンクの要件が満たされているかどうかを判定する
                リクエストがURL_LISTと一致する場合
                    // リンクがキューまたは収集されたセット内に存在するかどうかを確認します。そうでない場合は、キューに追加します。
                    if (!visited. contains (request) && !taskQueue. contains (request)) {
                        試す {
                            タスクキューにリクエストを書き込みます。
                        } キャッチ (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            })); 
 
        } キャッチ (例外 e) {
            e.printStackTrace();
        }
    }
    /**
     * ページを解析する
     *
     * @param URL
     */
    保護されたvoid processPage(文字列url) {
        試す {
            ドキュメント doc = Jsoup.connect (url).get();
            文字列タイトル = doc。 ( 「body > div.hp-wrap > div.voice-main > div.artical-title > h1」 )を選択します。最初に().ownText(); 
 
            システム。 out .println(Thread.currentThread().getName() + " in " + new Date () + " 収集されたHupuニュース " + title);
            // 収集したURLを収集セットに保存する
            訪問しました。追加(url); 
 
        } キャッチ (IOException e) {
            e.printStackTrace();
        }
    } 
 
公共 静的void main(String[] args) { 
 
        試す {
            新しい ThreadCrawler( "https://voice.hupu.com/nba" 、 5).run();
        } キャッチ (InterruptedException e) {
            e.printStackTrace();
        }
    }
 }

効果を確認するために、5 つのスレッドを使用して Hupu ニュースリストページを収集します。プログラムを実行すると、次の結果が得られます。

マルチスレッドコレクションの結果

結果からわかるように、61 ページを収集するために 5 つのスレッドを開始し、合計 2 秒かかりました。効果は依然として良好であると言えます。 1 本のスレッドと比較して、ギャップがどのくらい大きいか確認してみましょう。スレッド数を 1 に設定してプログラムを再度開始すると、次の結果が得られます。

シングルスレッド実行結果

シングルスレッドクローラーが Hupu から 61 件のニュースを収集するのに 7 秒かかったことがわかります。これは、マルチスレッドクローラーのほぼ 4 倍の時間です。考えてみてください、これはたった 61 ページです。ページ数が増えるとギャップはどんどん大きくなるため、マルチスレッドクローラーの効率は依然として非常に高くなります。

分散クローラーアーキテクチャ

分散クローラーアーキテクチャは、大規模な収集プログラムのみが使用する必要があるアーキテクチャです。一般的に、単一マシンのマルチスレッドはビジネスニーズを解決できます。いずれにせよ、私は分散クローラープロジェクトの経験がないので、これについては何も言うことはありません。しかし、技術者として、私たちは技術に対する熱意を持ち続ける必要があります。必要ではないですが、理解しておいても損はありません。私は多くの情報を調べ、以下の結論に達しました。

分散クローラーアーキテクチャは、考え方の面ではマルチスレッドクローラーアーキテクチャと同じです。シンプルな分散クローラーアーキテクチャにするには、マルチスレッドに基づいてわずかな改良を加えるだけで済みます。分散クローラーアーキテクチャではクローラーが異なるマシンに展開されるため、収集する URL と収集された URL をクローラーマシンのメモリに保存することはできません。 Redis や MongoDB などの特定のマシン上でそれらを管理する必要があります。各マシンは、LinkedBlockingQueue などのメモリキューからリンクを取得するのではなく、そこからコレクションリンクを取得します。このようにして、シンプルな分散クローラーアーキテクチャが実現します。もちろん、ここには多くの詳細があります。分散アーキテクチャの経験がないので、どこから始めればよいのかわかりません。ご興味がございましたら、ぜひご連絡ください。

<<: クラウドストレージの5つの利点

>>: クラウド上のフィンテック