编辑:为了澄清和简化:我正在寻找一种“好”的方式来在 Stackable 结束时向 Pool 提交更多 Stackable 对象(使用来自第一个 Stackable 的数据添加第二个一)。我的想法是轮询对象直到结束(低效且丑陋)并将引用传递给 Pool 对象(我无法使其工作)。 基本代码是这个:https://github.com/krakjoe/pthreads/blob/master/examples/Pooling.php

现在，完整的描述:

我正在使用 PHP 开发一个应用程序，该应用程序增长太多并且需要花费大量时间。因此，我正在尝试使用线程池对该应用程序进行多线程处理(我知道 PHP 不是最佳选择，但我不想，并且此时无法更改语言)。

问题是，应用程序有 2 个阶段，必须按顺序进行，每个阶段都有很多可以同时进行的子任务。所以，这是我脑海中的过程:

• 第 1 阶段将有 N 个子任务，这些子任务将是 Stackable 对象。
• 当子任务 i 结束时，必须通知“主要”(创建池、Stackable 等的那个)，并使用来自子任务 i 的一些数据(不同的 Stackable 对象)执行子任务 i 的阶段 2 ).在这个阶段，第 1 阶段的每个子任务都会有 M 个子任务。

我想为第 1 阶段和第 2 阶段的线程使用相同的线程池，从第 1 阶段到第 2 阶段我能想到的唯一解决方案是轮询 N 个子任务中的每一个，直到其中一个子任务结束，然后为结束的那个调用阶段 2，并重复直到所有 N 个子任务结束。

我使用 Joe Watkins 的 pthreads 源代码中包含的线程池示例作为基本代码。

最佳答案

您应该先阅读:https://gist.github.com/krakjoe/6437782

<?php
/**
* Normal worker
*/
class PooledWorker extends Worker {
    public function run(){}
}


/**
* Don't descend from pthreads, normal objects should be used for pools
*/
class Pool {
    protected $size;
    protected $workers;

    /**
    * Construct a worker pool of the given size
    * @param integer $size
    */  
    public function __construct($size) {
        $this->size = $size;
    }

    /**
    * Start worker threads
    */
    public function start() {
        while (@$worker++ < $this->size) {
            $this->workers[$worker] = new PooledWorker();
            $this->workers[$worker]->start();
        }
        return count($this->workers);
    }

    /**
    * Submit a task to pool
    */
    public function submit(Stackable $task) {
        $this->workers[array_rand($this->workers)]
            ->stack($task);
        return $task;
    }

    /**
    * Shutdown worker threads
    */
    public function shutdown() {
        foreach ($this->workers as $worker)
            $worker->shutdown();
    }
}

class StageTwo extends Stackable {
    /**
    * Construct StageTwo from a part of StageOne data
    * @param int $data
    */
    public function __construct($data) {
        $this->data = $data;
    }

    public function run(){
        printf(
            "Thread %lu got data: %d\n", 
            $this->worker->getThreadId(), $this->data);
    }
}

class StageOne extends Stackable {
    protected $done;

    /**
    * Construct StageOne with suitable storage for data
    * @param StagingData $data
    */
    public function __construct(StagingData $data) {
        $this->data = $data;
    }

    public function run() {
        /* create dummy data array */
        while (@$i++ < 100) {
            $this->data[] = mt_rand(
                20, 1000);
        }
        $this->done = true;
    }
}

/**
* StagingData to hold data from StageOne
*/
class StagingData extends Stackable {
    public function run() {}
}

/* stage and data reference arrays */
$one = [];
$two = [];
$data = [];

$pool = new Pool(8);
$pool->start();

/* construct stage one */
while (count($one) < 10) {
    $staging = new StagingData();
    /* maintain reference counts by storing return value in normal array in local scope */
    $one[] = $pool
        ->submit(new StageOne($staging));
    /* maintain reference counts */
    $data[] = $staging;
}

/* construct stage two */
while (count($one)) {

    /* find completed StageOne objects */
    foreach ($one as $id => $job) {
        /* if done is set, the data from this StageOne can be used */
        if ($job->done) {
            /* use each element of data to create new tasks for StageTwo */
            foreach ($job->data as $chunk) {
                /* submit stage two */
                $two[] = $pool
                    ->submit(new StageTwo($chunk));
            }

            /* no longer required */
            unset($one[$id]);
        }
    }

    /* in the real world, it is unecessary to keep polling the array */
    /* you probably have some work you want to do ... do it :) */
    if (count($one)) {
        /* everyone likes sleep ... */
        usleep(1000000);
    }
}

/* all tasks stacked, the pool can be shutdown */
$pool->shutdown();
?>

将输出:

Thread 140012266239744 got data: 612
Thread 140012275222272 got data: 267
Thread 140012257257216 got data: 971
Thread 140012033140480 got data: 881
Thread 140012257257216 got data: 1000
Thread 140012016355072 got data: 261
Thread 140012257257216 got data: 510
Thread 140012016355072 got data: 148
Thread 140012016355072 got data: 501
Thread 140012257257216 got data: 767
Thread 140012024747776 got data: 504
Thread 140012033140480 got data: 401
Thread 140012275222272 got data: 20
<-- trimmed from 1000 lines -->
Thread 140012041533184 got data: 285
Thread 140012275222272 got data: 811
Thread 140012041533184 got data: 436
Thread 140012257257216 got data: 977
Thread 140012033140480 got data: 830
Thread 140012275222272 got data: 554
Thread 140012024747776 got data: 704
Thread 140012033140480 got data: 50
Thread 140012257257216 got data: 794
Thread 140012024747776 got data: 724
Thread 140012033140480 got data: 624
Thread 140012266239744 got data: 756
Thread 140012284204800 got data: 997
Thread 140012266239744 got data: 708
Thread 140012266239744 got data: 981

因为您想使用一个池，所以您别无选择，只能在创建池的主上下文中创建任务。我可以想象其他解决方案，但您特别要求这种解决方案。

根据我可以使用的硬件以及要处理的任务和数据的性质，我可能有多个[小]线程池，每个工作线程一个，这将允许 StageOne 创建 StageTwo 对象在执行它们的 Worker 上下文中，这可能是需要考虑的事情。

关于PHP线程池，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/21098433/