PHP中的多线程、多卷曲爬虫

Question

大家好！

我们需要一些帮助来开发和实现多卷曲功能到我们的爬虫中。我们有大量“要扫描的链接”，我们用 Foreach 循环抛出它们。

让我们用一些伪代码来理解逻辑：

    1) While ($links_to_be_scanned > 0).
    2) Foreach ($links_to_be_scanned as $link_to_be_scanned).
    3) Scan_the_link() and run some other functions.
    4) Extract the new links from the xdom.
    5) Push the new links into $links_to_be_scanned.
    5) Push the current link into $links_already_scanned.
    6) Remove the current link from $links_to_be_scanned.

现在，我们需要定义最大并行连接数，并能够为每个链接并行运行此过程。

我知道我们将不得不创建一个 $links_being_scanned 或某种队列。

说实话，我真的不知道如何解决这个问题，如果有人能提供一些解决方案或想法，将不胜感激。

提前致谢！ 克里斯；

扩展：

我刚刚意识到，复杂的不是多卷曲本身，而是请求后每个链接完成的操作量。

即使在 muticurl 之后，我最终还是必须找到一种方法来并行运行所有这些操作。下面描述的整个算法必须并行运行。

所以现在重新考虑，我们将不得不做这样的事情：

  While (There's links to be scanned)
  Foreach ($Link_to_scann as $link)
  If (There's less than 10 scanners running)
  Launch_a_new_scanner($link)
  Remove the link from $links_to_be_scanned array
  Push the link into $links_on_queue array
  Endif;

每个扫描仪都会这样做（这应该并行运行）：

  Create an object with the given link
  Send a curl request to the given link
  Create a dom and an Xdom with the response body
  Perform other operations over the response body
  Remove the link from the $links_on_queue array
  Push the link into the $links_already_scanned array

我假设我们可以使用扫描仪算法创建一个新的 PHP 文件，并为每个并行进程使用 pcntl_fork()？

由于即使使用多卷曲，我最终还是不得不等待其他进程在常规 foreach 结构上循环。

我想我必须使用 fsockopen 或 pcntl_fork 来解决这个问题。

我们将不胜感激建议、cmets、部分解决方案，甚至是“祝你好运”！

非常感谢！

Answer 1

我将与您分享我用来从某个网站收集电子邮件地址的代码。 您可以对其进行修改以满足您的需要。 那里的相对 URL 存在一些问题。 而且我这里不使用 CURL。

<?php
error_reporting(E_ALL);
$home   = 'http://kharkov-reklama.com.ua/jborudovanie/';
$writer = new RWriter('C:\parser_13-09-2012_05.txt');
set_time_limit(0);
ini_set('memory_limit', '512M');

function scan_page($home, $full_url, &$writer) {

    static $done = array();
    $done[] = $full_url;

    // Scan only internal links. Do not scan all the internet!))
    if (strpos($full_url, $home) === false) {
        return false;
    }
    $html = @file_get_contents($full_url);
    if (empty($html) || (strpos($html, '<body') === false && strpos($html, '<BODY') === false)) {
        return false;
    }

    echo $full_url . '<br />';

    preg_match_all('/([A-Za-z0-9_\-]+\.)*[A-Za-z0-9_\-]+@([A-Za-z0-9][A-Za-z0-9\-]*[A-Za-z0-9]\.)+[A-Za-z]{2,4}/', $html, $emails);

    if (!empty($emails) && is_array($emails)) {
        foreach ($emails as $email_group) {
            if (is_array($email_group)) {
                foreach ($email_group as $email) {
                    if (filter_var($email, FILTER_VALIDATE_EMAIL)) {
                        $writer->write($email);
                    }
                }
            }
        }
    }

    $regexp = "<a\s[^>]*href=(\"??)([^\" >]*?)\\1[^>]*>(.*)<\/a>";
    preg_match_all("/$regexp/siU", $html, $matches, PREG_SET_ORDER);
    if (is_array($matches)) {
        foreach($matches as $match) {
            if (!empty($match[2]) && is_scalar($match[2])) {
                $url = $match[2];
                if (!filter_var($url, FILTER_VALIDATE_URL)) {
                    $url = $home . $url;
                }
                if (!in_array($url, $done)) {
                    scan_page($home, $url, $writer);
                }
            }
        }
    }
}

class RWriter {
    private $_fh = null;

    private $_written = array();

    public function __construct($fname) {
        $this->_fh = fopen($fname, 'w+');
    }

    public function write($line) {
        if (in_array($line, $this->_written)) {
            return;
        }
        $this->_written[] = $line;
        echo $line . '<br />';
        fwrite($this->_fh, "{$line}\r\n");
    }

    public function __destruct() {
        fclose($this->_fh);
    }
}

scan_page($home, 'http://kharkov-reklama.com.ua/jborudovanie/', $writer);

Answer 2

你可以试试这样的东西，还没有检查过，但你应该明白了

$request_pool = array();

function CreateHandle($url) {
    $handle = curl_init($url);

    // set curl options here

    return $handle;
}

function Process($data) {
    global $request_pool;

    // do something with data

    array_push($request_pool , CreateHandle($some_new_url));
}

function RunMulti() {
    global $request_pool;

    $multi_handle = curl_multi_init();

    $active_request_pool = array();

    $running = 0;
    $active_request_count = 0;
    $active_request_max = 10; // adjust as necessary
    do {
        $waiting_request_count = count($request_pool);
        while(($active_request_count < $active_request_max) && ($waiting_request_count > 0)) {
            $request = array_shift($request_pool);
            curl_multi_add_handle($multi_handle , $request);
            $active_request_pool[(int)$request] = $request;

            $waiting_request_count--;
            $active_request_count++;
        }

        curl_multi_exec($multi_handle , $running);
        curl_multi_select($multi_handle);
        while($info = curl_multi_info_read($multi_handle)) {
            $curl_handle = $info['handle'];
            call_user_func('Process' , curl_multi_getcontent($curl_handle));
            curl_multi_remove_handle($multi_handle , $curl_handle);
            curl_close($curl_handle);
            $active_request_count--;
        }

    } while($active_request_count > 0 || $waiting_request_count > 0);

    curl_multi_close($multi_handle);
}

Answer 3

您应该寻找一些更强大的解决方案来解决您的问题。 RabbitMQ 是我使用的一个非常好的解决方案。还有Gearman，但我认为这是你的选择。

我更喜欢 RabbitMQ。

Answer 4

免责声明：此答案链接了我参与的一个开源项目。那里。您已收到警告。

Artax HTTP client 是一个基于套接字的 HTTP 库，它（除其他外）在发出多个异步 HTTP 请求时提供对与各个主机的并发打开套接字连接数的自定义控制。

限制并发连接数很容易实现。考虑：

<?php

use Artax\Client, Artax\Response;

require dirname(__DIR__) . '/autoload.php';

$client = new Client;

// Defaults to max of 8 concurrent connections per host
$client->setOption('maxConnectionsPerHost', 2);

$requests = array(
    'so-home'    => 'http://stackoverflow.com',
    'so-php'     => 'http://stackoverflow.com/questions/tagged/php',
    'so-python'  => 'http://stackoverflow.com/questions/tagged/python',
    'so-http'    => 'http://stackoverflow.com/questions/tagged/http',
    'so-html'    => 'http://stackoverflow.com/questions/tagged/html',
    'so-css'     => 'http://stackoverflow.com/questions/tagged/css',
    'so-js'      => 'http://stackoverflow.com/questions/tagged/javascript'
);

$onResponse = function($requestKey, Response $r) {
    echo $requestKey, ' :: ', $r->getStatus();
};

$onError = function($requestKey, Exception $e) {
    echo $requestKey, ' :: ', $e->getMessage();
}

$client->requestMulti($requests, $onResponse, $onError);

重要提示：在上面的示例中，Client::requestMulti 方法异步发出所有指定的请求。因为每台主机的并发限制设置为2，客户端将为前两个请求打开新连接，随后为其他请求重用这些相同的套接字，将请求排队，直到两个套接字中的一个可用。