Во-первых, типы resource официально не поддерживаются pthreads; дескриптор curl - это resource, поэтому не следует хранить дескрипторы curl в области объектов объектов pthreads, так как они могут быть повреждены.

Упрощение

pthreads обеспечивает простой способ использования рабочих...

Самый простой способ выполнения среди множества потоков - использовать встроенный класс Pool, предоставляемый pthreads:

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);
?>

Ваша конкретная задача требует, чтобы вы сейчас обработали данные, вы можете либо записать эту функциональность в дизайн, подобный описанному выше... или

Делая это необычным

обещания - это супер причудливая форма параллелизма ...

Обещания соответствуют характеру задачи здесь:

• Сначала: Сделайте запрос
• Затем: Процесс ответ

Следующий код показывает, как использовать 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);
}
?>

Композитор:

{
    "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