php - 检测按顺序完成的线程

Question

我有以下多线程代码

class My_Thread extends Thread {

    public function run() {

        /* ... */
        // The time it takes to execut the code is different for each thread
    }
}

// Create a array
$threads = array();


/* *** STEP 1 *** */
//Initiate Miltiple Threads
foreach ( range("A", "B") as $i ) {
    $threads[] = new My_Thread($i);
}


/* *** STEP 2 *** */
// Start The Threads
foreach ($threads as $thread) {
    $thread->start(); // Thread A starts before thread B, and it takes more time to finish
}


/* *** STEP 3 *** */
// Process the threads
foreach ($threads as $thread) {
    if ($thread->join()) {
        /* ... Do Something ... */
    }
}

快速解释代码:

第 1 步:我正在创建两个线程，A 和 B

步骤 2:线程 A 先启动，完成时间比线程 B 长。

第 3 步:然后，我正在等待每个线程完成，从线程 A 开始。

现在，问题出在第 3 步。当我遍历线程时，我必须等待线程 A 完成才能进行进一步的处理，但是，线程 B 处于空闲状态等待，因为它需要更短的时间才能完成，除非在步骤3中处理线程A，否则不会处理。不保证线程A会花费更长的时间，所以我必须写一个通用的解决方案。

如何确保第 3 步处理首先完成的线程？换句话说，有没有类似这种伪代码的东西？

/* *** STEP 3 *** */
// Do the following for all threads in the $threads array, FIRST COME FIRST SERVE
// If the thread finished STEP 2, then immediately process it.

谢谢。

Answer 1

首先，一个Thread表示执行上下文。

您需要做的是分别考虑上下文和数据...

<?php
class Test extends Thread {

    public function __construct(Volatile $queue, $value) {
        $this->queue = $queue;
        $this->value = $value;
    }

    public function run() {
        $data = strlen(
            file_get_contents("http://www.google.co.uk/?q={$this->value}"));

        usleep(mt_rand(10000, 20000));

        $this->queue->synchronized(function($queue, $value, $data) {
            $queue[] = (array) [
                $value => $data
            ];
            $queue->notify();
        }, $this->queue, $this->value, $data);
    }

    private $queue;
    private $value;
}

$chars = ["A", "B", "C", "D", "E", "F", "G", "H", "I", "J"];
$queue = new Volatile();
$tests = [];

for ($test = 0; $test < 10; $test++) {
    $tests[$test] = new Test($queue, $chars[$test]);
    $tests[$test]->start();
}

$test = 0;

while (($next = $queue->synchronized(function() use($queue, &$test) {
    /* guard infinite loop */
    if (++$test > 10)
        return false;

    /* predicated wait for data */
    while (!count($queue))
        $queue->wait();

    /* return next item */
    return $queue->shift();
}))) {
    var_dump($next);
}

foreach ($tests as $thread)
    $thread->join();
?>

上面的代码适用于 pthreads v3、PHP7，它是可用的最佳版本，也是用于新项目的推荐版本。

解决方案的核心内容包含在 Test::run 中。和主上下文中的while循环。

$data = strlen(
    file_get_contents("http://www.google.co.uk/?q={$this->value}"));

usleep(mt_rand(10000, 20000));

这是为了从谷歌获取一些垃圾，碰巧响应时间是如此一致，以至于我不得不添加 usleep , 只是为了让您可以看到如果正确执行该顺序无关紧要。

你不应该使用 usleep在现实世界的多线程代码中。

$this->queue->synchronized(function($queue, $value, $data) {
    $queue[] = (array) [
        $value => $data
    ];
    $queue->notify();
}, $this->queue, $this->value, $data);

生成了一些数据后，每个 Test与队列同步，向其附加一些数据，并向当前等待的任何上下文发送通知。

与此同时，这正在发生:

$test = 0;

while (($next = $queue->synchronized(function() use($queue, &$test) {
    /* guard infinite loop */
    if (++$test > 10)
        return false;

    /* predicated wait for data */
    while (!count($queue))
        $queue->wait();

    /* return next item */
    return $queue->shift();
}))) {
    var_dump($next);
}

主上下文与队列同步，而在同步块(synchronized block)中它防止无限循环(因为我们知道有多少数据即将到来)，那么如果队列中没有数据，它将等待一些可用。最后将队列中的第一项返回到主上下文。

上面的代码将输出如下内容:

array(1) {
  ["I"]=>
  int(188965)
}
array(1) {
  ["B"]=>
  int(188977)
}
array(1) {
  ["C"]=>
  int(188921)
}
array(1) {
  ["F"]=>
  int(188962)
}
array(1) {
  ["J"]=>
  int(188954)
}
array(1) {
  ["A"]=>
  int(188912)
}
array(1) {
  ["E"]=>
  int(188929)
}
array(1) {
  ["G"]=>
  int(188941)
}
array(1) {
  ["D"]=>
  int(188946)
}
array(1) {
  ["H"]=>
  int(188929)
}

这里的关键是上下文和数据是不同的关注点。