PHP cURL - 线程安全吗？

Question

我编写了一个 PHP 脚本，它通过 libcurl 检索数据并对其进行处理。它工作正常，但出于性能原因，我将其更改为使用数十个工作线程(线程)。性能提高了 50 多倍，但是现在 php.exe 每隔几分钟就会崩溃，列出的错误模块是 php_curl.dll。我之前确实有过 C 语言的多线程经验，但之前在 php 中根本没有使用过它。

我用谷歌搜索了一下，据说 cURL 是线程安全的(截至 2001 年): http://curl.haxx.se/mail/lib-2001-01/0001.html但是我找不到任何关于 php_curl 是否线程安全的提及。

以防万一，我从命令行运行 php。我的设置是 Win7 x64、PHP 5.5.11 线程安全 VC11 x86、适用于 PHP 5.5 线程安全 VC11 x86 的 PHP pthreads 2.0.4。

这是一些伪代码来展示我在做什么

class MyWorker extends Worker
{
    ...
    public function run()
    {
        ...
        while(1)
        {
            ...
            runCURL();
            ...
            sleep(1);
        }
    }
}

function runCURL()
{
    static $curlHandle = null;
    ...
    if(is_null($curlHandle))
    {
        $curlHandle = curl_init();
        curl_setopt($curlHandle, CURLOPT_RETURNTRANSFER, TRUE);
        curl_setopt($curlHandle, CURLOPT_USERAGENT, "My User Agent String");
    }
    curl_setopt($curlHandle, CURLOPT_URL, "The URL");
    curl_setopt($curlHandle, CURLOPT_POSTFIELDS, $data);
    curl_setopt($curlHandle, CURLOPT_HTTPHEADER, $header);
    curl_setopt($curlHandle, CURLOPT_SSL_VERIFYPEER, false);

    $result = curl_exec($curlHandle);
    ...
}

Answer 1

首先，pthreads 官方不支持resource 类型； curl 句柄是一个资源，因此您不应将 curl 句柄存储在 pthreads 对象的对象范围内，因为它们可能会损坏。

让它变得简单

pthreads 提供了一种使用 worker 的简单方法...

在多个线程之间执行的最简单方法是使用 pthreads 提供的内置 Pool 类:

http://php.net/pool

下面的代码演示了如何在几个后台线程中合并一堆请求:

<?php
define("LOG", Mutex::create());

function slog($message, $args = []) {
    $args = func_get_args();
    if (($message = array_shift($args))) {
        Mutex::lock(LOG);
        echo vsprintf("{$message}\n", $args);
        Mutex::unlock(LOG);
    }
}

class Request extends Threaded {

    public function __construct($url, $post = []) {
        $this->url = $url;
        $this->post = $post;
    }

    public function run() {
        $curl = curl_init();

        curl_setopt($curl, CURLOPT_RETURNTRANSFER, true);
        curl_setopt($curl, CURLOPT_URL, $this->url);

        if ($this->post) {
            curl_setopt($curl, CURLOPT_POSTFIELDS, $this->post);
        }

        $response = curl_exec($curl);

        slog("%s returned %d bytes", $this->url, strlen($response));
    }

    public function getURL()      { return $this->url;      }
    public function getPost()     { return $this->post;     }

    protected $url;
    protected $post;
}

$max = 100;
$urls   = [];
while (count($urls) < $max) {
    $urls[] = sprintf(
        "http://www.google.co.uk/?q=%s", 
        md5(mt_rand()*count($urls)));
}

$pool = new Pool(4);

foreach ($urls as $url) {
    $pool->submit(new Request($url));
}

$pool->shutdown();

Mutex::destroy(LOG);
?>

您的特定任务要求您现在处理数据，您可以将此功能写入上述设计中......或者

花哨的

promises 是一种 super 奇特的并发形式......

Promise 适合此处任务的性质:

• 首先:提出请求
• 然后:处理响应

以下代码显示了如何使用pthreads/promises 发出相同的请求并处理响应:

<?php
namespace {
    require_once("vendor/autoload.php");

    use pthreads\PromiseManager;
    use pthreads\Promise;
    use pthreads\Promisable;
    use pthreads\Thenable;

    define("LOG", Mutex::create());

    function slog($message, $args = []) {
        $args = func_get_args();
        if (($message = array_shift($args))) {
            Mutex::lock(LOG);
            echo vsprintf("{$message}\n", $args);
            Mutex::unlock(LOG);
        }
    }

    /* will be used by everything to report errors when they occur */
    trait ErrorManager {
        public function onError(Promisable $promised) {
            slog("Oh noes: %s\n", (string) $promised->getError());
        }
    }

    class Request extends Promisable {
        use ErrorManager;

        public function __construct($url, $post = []) {
            $this->url = $url;
            $this->post = $post;
            $this->done = false;
        }

        public function onFulfill() {
            $curl = curl_init();

            curl_setopt($curl, CURLOPT_RETURNTRANSFER, true);
            curl_setopt($curl, CURLOPT_URL, $this->url);

            if ($this->post) {
                curl_setopt($curl, CURLOPT_POSTFIELDS, $this->post);
            }

            $this->response = curl_exec($curl);
        }

        public function getURL()      { return $this->url;      }
        public function getPost()     { return $this->post;     }
        public function getResponse() { return $this->response; }
        public function setGarbage()  { $this->garbage = true; }
        public function isGarbage()   { return $this->garbage; }

        protected $url;
        protected $post;
        protected $response;
        protected $garbage;
    }

    class Process extends Thenable {
        use ErrorManager;

        public function onFulfilled(Promisable $request) {
            slog("%s returned %d bytes\n",  
                $request->getURL(), strlen($request->getResponse()));
        }
    }

    /* some dummy urls */
    $max = 100;
    $urls   = [];
    while (count($urls) < $max) {
        $urls[] = sprintf(
            "http://www.google.co.uk/?q=%s", 
            md5(mt_rand()*count($urls)));
    }

    /* initialize manager for promises */
    $manager = new PromiseManager(4);

    /* create promises to make and process requests */
    while (@++$id < $max) {
        $promise = new Promise($manager, new Request($urls[$id], []));
        $promise->then(
            new Process($promise));
    }

    /* force the manager to shutdown (fulfilling all promises first) */
    $manager->shutdown();

    /* destroy mutex */
    Mutex::destroy(LOG);
}
?>

Composer :

{
    "require": {
        "krakjoe/promises": ">=1.0.2"
    }
}

请注意，Request 几乎没有变化，添加的只是保存响应的地方以及检测对象是否为垃圾的方法。

有关从池中收集垃圾的详细信息，这适用于两个示例:

http://php.net/pool.collect

slog 函数的存在只是为了使记录的输出可读

说清楚

pthreads 不是新的 PDO 驱动程序......

许多人使用 pthreads 就像他们使用新的 PDO 驱动程序一样 - 假设它像 PHP 的其余部分一样工作并且一切都会很好。

一切可能都不是很好，需要研究:我们正在挑战极限，在这样做的过程中，必须对 pthreads 的体系结构施加一些“限制”以保持稳定性，这可能会有一些奇怪的副作用。

虽然 pthreads 附带详尽的文档，其中大部分包括 PHP 手册中的示例，但我无法在手册中附加以下文档。

以下文档让您了解 pthreads 的内部结构，每个人都应该阅读它，它是为您编写的。

https://gist.github.com/krakjoe/6437782