pthreads销毁数组结果

Question

我有一个使用pthreadsv3.1.6和PHP 7.0.22的代码。我遇到的问题是线程不返回数组值。我的代码如下

$threadCount = 1;

$url_idList = range(1,2000);
$url_idListChunked = array_chunk($url_idList, $threadCount);

class WorkerThreads extends Thread {
    private $threadName, $url_id;

    public function __construct($threadName,$url_id) {
        $this->threadName = $threadName;
        $this->url_id = $url_id;
        $this->result = [];
    }

    public function run() {
        if ($this->threadName && $this->url_id) {
            printf('%sLoading URL #: %s' . "\n", $this->threadName, $this->url_id);
            $this->result = send_request('GET',$this->url_id,NULL,$this->threadName);
        }
    }
}

while(count($url_idListChunked)){
    $url_idListChunk = array_shift($url_idListChunked);
    $workers = [];
    foreach (range(0,count($url_idListChunk)-1) as $i) {
        $threadName = "Thread #".$i.": ";
        $workers[$i] = new WorkerThreads($threadName,$url_idListChunk[$i]);
        $workers[$i]->start();
    }

    foreach (range(0,count($url_idListChunk)-1) as $i) {
        $workers[$i]->join();
        print_r($workers[$i]);
        exit();
        echo $workers[$i]['threadName']."Result for URL #: ".$workers[$i]['url_id']."\n";
    }

}

function send_request($method,$url_id,$data,$threadName=NULL){

    $url = 'https://www.example.tld/?id='.$url_id;

    $ch = curl_init();
    curl_setopt($ch, CURLOPT_URL, $url);
    curl_setopt($ch, CURLOPT_HEADER, TRUE);
    curl_setopt($ch, CURLOPT_RETURNTRANSFER, TRUE);
    curl_setopt($ch, CURLOPT_FOLLOWLOCATION, TRUE);
    curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, FALSE);
    curl_setopt($ch, CURLOPT_TIMEOUT, 3);
    if(!$data && $method=='POST'){
        $data = generate_post_data();
        curl_setopt($ch, CURLOPT_POST, 1);
        curl_setopt($ch, CURLOPT_POSTFIELDS, $data);
    }
    $response = curl_exec($ch);
    while((curl_errno($ch) == 6 OR curl_errno($ch) == 28)){
        $response = curl_exec($ch);
        echo $threadName.'Curl error #'.curl_errno($ch).' - ' . curl_error($ch)." Retrying.\n";
        sleep(2);       
    }

    curl_close($ch);

    $result['data'] = $response;

    return $result;

}

当我得到try to print_t($workers)时，我得到了以下错误消息Uncaught RuntimeException: pthreads detected an attempt to connect to an object which has already been destroyed。为什么我会丢失数组结果？看起来线程传递回字符串没有任何问题。

Answer 1

Hie，我不完全确定使用start->join方法是实现您想要实现的目标的首选方法？我认为您需要使用pthread中为其创建的池可收集方法。

如果您想从分块的批处理中收集结果，这里有一个示例可能会在您的工作中启发您。我个人经常使用它，这是将pthread推向极限的最快方法。注意不要将池号推送到您的CPU线程上(在本例中，它是10个核心)。

如果我可以说，关于你的代码，永远不要尝试在屏幕上输出来自pthread worker的东西，它肯定会乱来。返回对象并在集合上回显它。确保你的类中有公共的结果，以允许返回的对象。

更不用说多卷曲，这可能是最快和最合适的方式。

/* pthreads batches */
$batches = array();

$nbpool = 20; // cpu 10 cores 

/* job 1 */
$list = [/* data1 */];
$url_idList[] = array_chunk($list, 5000);

/* job 2 */
$list2 = [/* data2 */];
$url_idList[] = array_chunk($list, 10000);

/* final collected results */
$resultFinal = [];

/* loop across batches */
foreach ($url_idList as $key => $url_idListChunked) {

    $url_idListChunk = array_shift($url_idListChunked);

    /* for intermediate collection */
    $data[$key] = [];

    /* how many workers */
    $workCount = count($url_idListChunk);

    /* set pool job up to max cpu capabilities */
    $pool = new Pool($nbpool, Worker::class);

    /* pool cycling submit */
    foreach (range(1, $workCount) as $i) {
        $chunck = $url_idListChunk[$i - 1];
        $pool->submit(new WorkerThreads(($i - 1), $chunck));
    }

    /* on collection cycling */
    $collector = function (\Collectable $work) use (&$data) {

        /* is worker complete ? */
        $isGarbage = $work->isGarbage();

        /* worker complete */
        if ($isGarbage) {
            $result = $work->result;
            $info = $work->info;
            $data[$key] = $result;

            /* echo result info outside worker */
            echo($info);
        }
        return $isGarbage;
    };
    do {
        /* collection on pool stack */
        $count = $pool->collect($collector);
        $isComplete = count($data) === $workCount;
    } while (!$isComplete);

    /* push stack results */
    array_push($resultFinal, $data);

    /* close pool */
    $pool->shutdown();
}

class WorkerThreads extends \Threaded implements \Collectable {

    private $url_id;
    private $isGarbage;
    private $threadName;
    public $result;
    public $info;

    public function __construct($i, $url_id) {
        $this->threadName = "Thread #" . $i . ": ";
        $this->url_id = $url_id;
    }

    public function run() {
        if ($this->threadName && $this->url_id) {
            $this->result = send_request('GET', $this->url_id, NULL, $this->threadName);
        }
        $this->info = $this->threadName . " Result for URL #: " . $this->url_id;
        $this->isGarbage = true; // yeah, it s done
    }

    public function isGarbage(): bool {
        return $this->isGarbage;
    }

}