怎么在PHP中使用worker和pool

今天就跟大家聊聊有關(guān)怎么在PHP中使用worker和pool，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

<?php
//創(chuàng)建自定義work類(lèi)，給work取個(gè)名字，方便查看
class Work extends Worker
{
  private $name;

  public function __construct($name)
  {
    $this->name = $name;
  }

  public function getName()
  {
    return $this->name;
  }
}

class Task extends Thread
{
  private $num;

  public function __construct($num)
  {
    $this->num = $num;
  }

  public function run()
  {
    //計(jì)算累加和
    $total = 0;
    for ($i = 0; $i < $this->num; $i++) {
      $total += $i;
    }
    echo "work : {$this->worker->getName()} task : {$total} \n";
    sleep(1);
  }
}

//創(chuàng)建一個(gè)worker線(xiàn)程
$work = new Work('a');

$work->start();

for ($i = 1; $i <= 10; $i++) {
  //將Task對(duì)象壓棧到worker線(xiàn)程中
  //這個(gè)時(shí)候Task對(duì)象就可以使用worker線(xiàn)程上下文(變量，函數(shù)等)
  $work->stack(new Task($i));
}

//循環(huán)的清理任務(wù)，會(huì)阻塞主線(xiàn)程，直到棧中任務(wù)都執(zhí)行完畢
while ($work->collect()) ;

//關(guān)閉worker
$work->shutdown();

上面代碼在運(yùn)行的時(shí)候，計(jì)算結(jié)果會(huì)每隔一秒出來(lái)一條，也就是10個(gè)task對(duì)象是運(yùn)行在1個(gè)worker線(xiàn)程上的。

如果10個(gè)task對(duì)象是分別在獨(dú)立空間運(yùn)行的，sleep()函數(shù)就不會(huì)起作用，他們各自sleep并不會(huì)影響其他線(xiàn)程。

把上面的代碼修改一下：

<?php
//創(chuàng)建自定義work類(lèi)，給work取個(gè)名字，方便查看
class Work extends Worker
{
  private $name;

  public function __construct($name)
  {
    $this->name = $name;
  }

  public function getName()
  {
    return $this->name;
  }
}

class Task extends Thread
{
  private $num;

  public function __construct($num)
  {
    $this->num = $num;
  }

  public function run()
  {
    //計(jì)算累加和
    $total = 0;
    for ($i = 0; $i < $this->num; $i++) {
      $total += $i;
    }
    echo "work : {$this->worker->getName()} task : {$total} \n";
    sleep(1);
  }
}

//創(chuàng)建二個(gè)worker線(xiàn)程
$work1 = new Work('a');
$work2 = new Work('b');

$work1->start();
$work2->start();

for ($i = 1; $i <= 10; $i++) {
  if ($i <= 5) {
    $work1->stack(new Task($i));
  } else {
    $work2->stack(new Task($i));
  }
}

//循環(huán)的清理任務(wù)，會(huì)阻塞主線(xiàn)程，直到棧中任務(wù)都執(zhí)行完畢
while ($work1->collect() || $work2->collect()) ;

//關(guān)閉worker
$work1->shutdown();
$work2->shutdown();

這里我們創(chuàng)建2個(gè)worker線(xiàn)程，讓10個(gè)task對(duì)象分別壓棧到2個(gè)worker中。

這時(shí)可以看到，計(jì)算結(jié)果是一對(duì)一對(duì)的出來(lái)，說(shuō)明10個(gè)task對(duì)象跑在了2個(gè)worker線(xiàn)程上。

至于需要?jiǎng)?chuàng)建多少個(gè)worker線(xiàn)程，和多少個(gè)task對(duì)象，就看自已的需求了。

worker還有一個(gè)好處就是可以重用worker中的對(duì)象和方法。我們可以在worker中創(chuàng)建一個(gè)連接數(shù)據(jù)庫(kù)對(duì)象，方便各task調(diào)用。

<?php
class DB extends Worker
{
  //注意這里設(shè)置為靜態(tài)成員，pdo連接本身是不能在上下文中共享的
  //聲明為靜態(tài)成員，讓每個(gè)worker有自已的pdo連接
  private static $db = null;
  public $msg = 'i from db';

  public function run()
  {
    self::$db = new PDO('mysql:host=192.168.33.226;port=3306;dbname=test;charset=utf8', 'root', '');
  }

  public function getDb()
  {
    return self::$db;
  }
}

class Task extends Thread
{
  private $id;
  //注意，這里不要給成員設(shè)置默認(rèn)值，$result成員是線(xiàn)程對(duì)象是不可變的，不能被改寫(xiě)
  private $result;

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

  public function run()
  {
    //獲取worker中的數(shù)據(jù)庫(kù)連接
    $db = $this->worker->getDb();
    $ret = $db->query("select * from tb_user where id = {$this->id}");
    $this->result = $ret->fetch(PDO::FETCH_ASSOC);
    //訪問(wèn)worker中的成員變量msg
    echo "data : {$this->result['id']} {$this->result['name']} \t worker data : {$this->worker->msg} \n";
  }
}

//創(chuàng)建一個(gè)worker線(xiàn)程
$work = new DB();

$work->start();

for ($i = 1; $i <= 5; $i++) {
  $work->stack(new Task($i));
}

//循環(huán)的清理任務(wù)，會(huì)阻塞主線(xiàn)程，直到棧中任務(wù)都執(zhí)行完畢
while ($work->collect()) ;

//關(guān)閉worker
$work->shutdown();

tb_user表大家可以隨意創(chuàng)建，我這里為了演示只創(chuàng)建了id和name字段

運(yùn)行結(jié)果如下：

如果說(shuō)worker是對(duì)線(xiàn)程的重用，那么pool就是對(duì)worker更高的抽象了，可以同時(shí)管理多個(gè)worker。

<?php
//之所以要?jiǎng)?chuàng)建一個(gè)Id線(xiàn)程類(lèi)，主要是為了給work取個(gè)不同的ID，方便查看，哪些task線(xiàn)程屬于哪個(gè)work中
class Id extends Thread
{
  private $id;

  public function getId()
  {
    //防止出現(xiàn)id混亂，這里使用同步操作
    $this->synchronized(function () {
      ++$this->id;
    });
    return $this->id;
  }
}

class Work extends Worker
{
  private $id;

  public function __construct(Id $obj)
  {
    $this->id = $obj->getId();
  }

  public function getId()
  {
    return $this->id;
  }
}

class Task extends Thread
{
  private $num = 0;

  public function __construct($num)
  {
    $this->num = $num;
  }

  //計(jì)算累加和
  public function run()
  {
    $total = 0;
    for ($i = 0; $i < $this->num; $i++) {
      $total += $i;
    }
    echo "work id : {$this->worker->getId()} task : {$total} \n";
  }
}

//創(chuàng)建pool，可容納3個(gè)work對(duì)象
$pool = new Pool(3, 'Work', [new Id()]);

//循環(huán)的把20個(gè)task線(xiàn)程提交到pool中的work對(duì)象上運(yùn)行
for ($i = 1; $i <= 20; $i++) {
  $pool->submit(new Task($i));
}

//循環(huán)的清理任務(wù)，會(huì)阻塞主線(xiàn)程，直到任務(wù)都執(zhí)行完畢
while ($pool->collect()) ;

//關(guān)閉pool
$pool->shutdown();

運(yùn)行結(jié)果如下：

看完上述內(nèi)容，你們對(duì)怎么在PHP中使用worker和pool有進(jìn)一步的了解嗎？如果還想了解更多知識(shí)或者相關(guān)內(nèi)容，請(qǐng)關(guān)注億速云行業(yè)資訊頻道，感謝大家的支持。