为什么不是所有线程都完成？

Question

我已经尝试过这个 Joe 回答/sf/answers/2253097241/ 中的示例，它运行良好，但是当我尝试稍微编辑此代码时：

$pool = new Pool(4);

while (@$i++<10) {
    $pool->submit(new class($i) extends Collectable {
        public function __construct($id) {
            $this->id = $id;
        }

        public function run() {
            printf(
                "Hello World from %d\n", $this->id);
        $this->html = file_get_contents('http://google.fr?q=' . $this->query);
            $this->setGarbage();
        }

        public $id;
public $html;
    });
}

while ($pool->collect(function(Collectable $work){
    printf(
        "Collecting %d\n", $work->id);
var_dump($work->html);
    return $work->isGarbage();
})) continue;

$pool->shutdown();

“Hello world”的计数与“Collecting”的计数不同。文档已过时。这个问题怎么办？

Answer 1

Worker::collect并非旨在让您获得成果；它是不确定的。

Worker::collect仅用于对对象堆栈中引用的对象运行垃圾收集Worker。

如果目的是在每个结果可用时对其进行处理，则代码可能如下所示：

<?php
$pool = new Pool(4);
$results = new Volatile();
$expected = 10;
$found = 0;

while (@$i++ < $expected) {
    $pool->submit(new class($i, $results) extends Threaded {

        public function __construct($id, Volatile $results) {
            $this->id = $id;
            $this->results = $results;
        }

        public function run() {
            $result = file_get_contents('http://google.fr?q=' . $this->id);

            $this->results->synchronized(function($results, $result){
                $results[$this->id] = $result;
                $results->notify();
            }, $this->results, $result);
        }

        private $id;
        private $results;
    });
}

do {
    $next = $results->synchronized(function() use(&$found, $results) {
        while (!count($results)) {
            $results->wait();
        }

        $found++;

        return $results->shift();
    });

    var_dump($next);
} while ($found < $expected);

while ($pool->collect()) continue;

$pool->shutdown();
?>

这显然不太容错，但主要区别是我使用共享Volatile的结果集合，并且当结果可用时我会正确同步以在主上下文中获取结果。

如果您想等待所有结果都可用，并且可能避免一些锁争用（如果可以的话，您应该始终尽量避免这种争用），那么代码会看起来更简单，如下所示：

<?php
$pool = new Pool(4);
$results = new Volatile();
$expected = 10;

while (@$i++ < $expected) {
    $pool->submit(new class($i, $results) extends Threaded {

        public function __construct($id, Volatile $results) {
            $this->id = $id;
            $this->results = $results;
        }

        public function run() {
            $result = file_get_contents('http://google.fr?q=' . $this->id);

            $this->results->synchronized(function($results, $result){
                $results[$this->id] = $result;
                $results->notify();
            }, $this->results, $result);
        }

        private $id;
        private $results;
    });
}

$results->synchronized(function() use($expected, $results) {
    while (count($results) != $expected) {
        $results->wait();
    }
});

var_dump(count($results));

while ($pool->collect()) continue;

$pool->shutdown();
?>

值得注意的是，该Collectable接口已经Threaded在最新版本的 pthreads 中实现 - 这是您应该使用的......总是......

这些文档已经过时了，对此感到抱歉......一个人......