怎么在PHP中使用worker和pool

tangjin

作者

今天就跟大家聊聊有关怎么在PHP中使用worker和pool，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。

<?php
//创建自定义work类，给work取个名字，方便查看
classWorkextendsWorker
{
private$name;

publicfunction__construct($name)
{
$this->name=$name;
}

publicfunctiongetName()
{
return$this->name;
}
}

classTaskextendsThread
{
private$num;

publicfunction__construct($num)
{
$this->num=$num;
}

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

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

$work->start();

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

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

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

上面代码在运行的时候，计算结果会每隔一秒出来一条，也就是10个task对象是运行在1个worker线程上的。

如果10个task对象是分别在独立空间运行的，sleep()函数就不会起作用，他们各自sleep并不会影响其他线程。

把上面的代码修改一下：

<?php
//创建自定义work类，给work取个名字，方便查看
classWorkextendsWorker
{
private$name;

publicfunction__construct($name)
{
$this->name=$name;
}

publicfunctiongetName()
{
return$this->name;
}
}

classTaskextendsThread
{
private$num;

publicfunction__construct($num)
{
$this->num=$num;
}

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

//创建二个worker线程
$work1=newWork('a');
$work2=newWork('b');

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

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

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

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

这里我们创建2个worker线程，让10个task对象分别压栈到2个worker中。

这时可以看到，计算结果是一对一对的出来，说明10个task对象跑在了2个worker线程上。

至于需要创建多少个worker线程，和多少个task对象，就看自已的需求了。

worker还有一个好处就是可以重用worker中的对象和方法。我们可以在worker中创建一个连接数据库对象，方便各task调用。

<?php
classDBextendsWorker
{
//注意这里设置为静态成员，pdo连接本身是不能在上下文中共享的
//声明为静态成员，让每个worker有自已的pdo连接
privatestatic$db=null;
public$msg='ifromdb';

publicfunctionrun()
{
self::$db=newPDO('mysql:host=192.168.33.226;port=3306;dbname=test;charset=utf8','root','');
}

publicfunctiongetDb()
{
returnself::$db;
}
}

classTaskextendsThread
{
private$id;
//注意，这里不要给成员设置默认值，$result成员是线程对象是不可变的，不能被改写
private$result;

publicfunction__construct($id)
{
$this->id=$id;
}

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

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

$work->start();

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

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

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

tb_user表大家可以随意创建，我这里为了演示只创建了id和name字段

运行结果如下：

如果说worker是对线程的重用，那么pool就是对worker更高的抽象了，可以同时管理多个worker。

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

publicfunctiongetId()
{
//防止出现id混乱，这里使用同步操作
$this->synchronized(function(){
++$this->id;
});
return$this->id;
}
}

classWorkextendsWorker
{
private$id;

publicfunction__construct(Id$obj)
{
$this->id=$obj->getId();
}

publicfunctiongetId()
{
return$this->id;
}
}

classTaskextendsThread
{
private$num=0;

publicfunction__construct($num)
{
$this->num=$num;
}

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

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

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

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

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

运行结果如下：