PHP pthreads v3下worker和pool的使用方法示例

<?php
//创建自定义work类，给work取个名字，方便查看
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()
  {
    //计算累加和
    $total = 0;
    for ($i = 0; $i < $this->num; $i++) {
      $total += $i;
    }
    echo "work : {$this->worker->getName()} task : {$total} \n";
    sleep(1);
  }
}

//创建一个worker线程
$work = new Work('a');

$work->start();

for ($i = 1; $i <= 10; $i++) {
  //将Task对象压栈到worker线程中
  //这个时候Task对象就可以使用worker线程上下文(变量，函数等)
  $work->stack(new Task($i));
}

//循环的清理任务，会阻塞主线程，直到栈中任务都执行完毕
while ($work->collect()) ;

//关闭worker
$work->shutdown();

<?php
//创建自定义work类，给work取个名字，方便查看
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()
  {
    //计算累加和
    $total = 0;
    for ($i = 0; $i < $this->num; $i++) {
      $total += $i;
    }
    echo "work : {$this->worker->getName()} task : {$total} \n";
    sleep(1);
  }
}

//创建二个worker线程
$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));
  }
}

//循环的清理任务，会阻塞主线程，直到栈中任务都执行完毕
while ($work1->collect() || $work2->collect()) ;

//关闭worker
$work1->shutdown();
$work2->shutdown();

<?php
class DB extends Worker
{
  //注意这里设置为静态成员，pdo连接本身是不能在上下文中共享的
  //声明为静态成员，让每个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;
  //注意，这里不要给成员设置默认值，$result成员是线程对象是不可变的，不能被改写
  private $result;

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

  public function run()
  {
    //获取worker中的数据库连接
    $db = $this->worker->getDb();
    $ret = $db->query("select * from tb_user where id = {$this->id}");
    $this->result = $ret->fetch(PDO::FETCH_ASSOC);
    //访问worker中的成员变量msg
    echo "data : {$this->result['id']} {$this->result['name']} \t worker data : {$this->worker->msg} \n";
  }
}

//创建一个worker线程
$work = new DB();

$work->start();

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

//循环的清理任务，会阻塞主线程，直到栈中任务都执行完毕
while ($work->collect()) ;

//关闭worker
$work->shutdown();

<?php
//之所以要创建一个Id线程类，主要是为了给work取个不同的ID，方便查看，哪些task线程属于哪个work中
class Id extends Thread
{
  private $id;

  public function getId()
  {
    //防止出现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;
  }

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

//创建pool，可容纳3个work对象
$pool = new Pool(3, 'Work', [new Id()]);

//循环的把20个task线程提交到pool中的work对象上运行
for ($i = 1; $i <= 20; $i++) {
  $pool->submit(new Task($i));
}

//循环的清理任务，会阻塞主线程，直到任务都执行完毕
while ($pool->collect()) ;

//关闭pool
$pool->shutdown();