这个问题遵循这个答案：为什么不是所有的线程都完成了？
我试图将乔·沃特金斯的示例实现到几个类中：
SynchronizedWork ：
<?php
class SynchronizedWork extends Threaded {
/**
* @var int Dummy work ID
*/
protected $id;
/**
* @var $resultBag Volatile Result container passed around threads
*/
protected $resultBag;
/**
* @var bool Should this work be collected for garbage ?
*/
protected $garbage = 0;
public function __construct($id)
{
$this->id = $id;
}
public function getId()
{
return $this->id;
}
public function setResultBag(Volatile $volatile)
{
$this->resultBag = $volatile;
return $this;
}
/**
* @return $this Set this work ready to be collected for garbage
*/
public function setGarbage()
{
$this->garbage = 1;
return $this;
}
/**
* @return bool Is this work ready to be collected for garbage ?
*/
public function isGarbage() : bool
{
return $this->garbage === 1;
}
public function run()
{
$this->send(range(0, 10, $this->getId()));
}
public function send($result)
{
$this->resultBag->synchronized(function(Volatile $results, $workResult){
$results[$this->getId()] = $workResult;
$results->notify();
}, $this->resultBag, $result);
}
}
SynchronizedPool ：
class SynchronizedPool extends Pool {
/**
* @var int Pool's lifetime submitted works count
*/
protected $submitted = 0;
/**
* @var int Pool's lifetime collected works count
*/
protected $collected = 0;
protected $resultBag;
public function __construct($size, $clazz = Worker::class, $ctor = array())
{
$this->resultBag = new Volatile;
parent::__construct($size, $clazz, $ctor);
}
/**
* @param Threaded $threaded Submit the work to the pool and will be dispatched among Pool's Workers
* @return int|void
*/
public function submit(Threaded $threaded)
{
++$this->submitted;
$threaded->setResultBag($this->resultBag);
parent::submit($threaded);
}
public function getResultBag()
{
return $this->resultBag;
}
/**
* @return int Return the number of work submitted since the Pool has been created
*/
public function submitted()
{
return $this->submitted;
}
/**
* @return array Return the list of workers in the stack. Collected workers are no longer in this stack.
*/
public function getWorks() : array
{
return $this->workers;
}
/**
* @return int Return the number of work collected for garbage since the Pool has been created
*/
public function collected()
{
return $this->collected;
}
public function processResult(Closure $closure)
{
$found = 0;
$resultBag = $this->getResultBag();
do {
$workResult = $this->getResultBag()->synchronized(function() use(&$found, $resultBag) {
while (!count($resultBag)) {
$resultBag->wait();
}
$found++;
return $resultBag->shift();
});
$closure($workResult);
} while ($found < $this->submitted());
}
}
和执行：
$pool = new SynchronizedPool(3);
$pool->submit(new SynchronizedWork(1));
$pool->submit(new SynchronizedWork(2));
$pool->submit(new SynchronizedWork(3));
$pool->processResult(function($workResult) {
var_dump($workResult);
});
while($pool->collect()) continue;
$pool->shutdown();
它似乎失败，并显示：
PHP Fatal error: Uncaught RuntimeException: pthreads detected an attempt to connect to an object which has already been destroyed in /mnt/hgfs/IdeaProjects/backoffice-data/web/pthreads.php:137
Stack trace:
#0 /mnt/hgfs/IdeaProjects/backoffice-data/web/pthreads.php(137): Threaded->shift()
#1 [internal function]: SynchronizedPool->{closure}()
#2 /mnt/hgfs/IdeaProjects/backoffice-data/web/pthreads.php(138): Threaded->synchronized(Object(Closure))
#3 /mnt/hgfs/IdeaProjects/backoffice-data/web/pthreads.php(154): SynchronizedPool->processResult(Object(Closure))
#4 {main}
thrown in /mnt/hgfs/IdeaProjects/backoffice-data/web/pthreads.php on line 137
我还尝试在提交之前调用 processResult，认为也许 ref 在结束后立即被销毁（即使我没有调用 collect），但在这种情况下，它只是永远等待。
为什么说结果包被销毁了？裁判仍然在游泳池中持有，不是吗？我做错了什么/理解错了什么？
在抓取了 pthreads 的 github 和一些 SO 问题后，我发现了这个问题，并揭穿了下一个问题：
在 send 方法中，我必须将线程的结果强制转换为实际数组，以防止它被强制转换为将在方法结束时被销毁的Volatile（如果我理解正确的话）。这导致：
在SynchronizedWork课上。
public function send($result)
{
$this->resultBag->synchronized(function(Volatile $resultBag, $workResult){
$resultBag[$this->getId()] = (array) $workResult;
$resultBag->notify();
}, $this->resultBag, $result);
}
然后，将线程设置为垃圾回收，否则池收集将永远等待。
public function run()
{
$holdenRef = range(0, 10, $this->getId());
$this->send($holdenRef);
$this->setGarbage();
}