本文实例讲述了PHP pthreads v3下worker和pool的使用方法，分享给大家供大家参考，具体如下：

有些人会想，明明用thread已经可以很好的工作了，为什么还要搞个worker和pool？

之所以要用到worker和pool还是因为效率，因为系统创建一个新线程代价是比较昂贵，每个创建的线程会复制当前执行的整个上下文。

尽可能的重用线程可以让我们的程序更高效。

一个简单的worker例子：

1. <?php 
2. //创建自定义work类，给work取个名字，方便查看 
3. class Work extends Worker 
4. { 
5.   private $name; 
6.  
7.   public function __construct($name) 
8.   { 
9.     $this->name = $name; 
10.   } 
11.  
12.   public function getName() 
13.   { 
14.     return $this->name; 
15.   } 
16. } 
17.  
18. class Task extends Thread 
19. { 
20.   private $num; 
21.  
22.   public function __construct($num) 
23.   { 
24.     $this->num = $num; 
25.   } 
26.  
27.   public function run() 
28.   { 
29.     //计算累加和 
30.     $total = 0; 
31.     for ($i = 0; $i < $this->num; $i++) { 
32.       $total += $i; 
33.     } 
34.     echo "work : {$this->worker->getName()} task : {$total} \n"; 
35.     sleep(1); 
36.   } 
37. } 
38.  
39. //创建一个worker线程 
40. $work = new Work('a'); 
41.  
42. $work->start(); 
43.  
44. for ($i = 1; $i <= 10; $i++) { 
45.   //将Task对象压栈到worker线程中 
46.   //这个时候Task对象就可以使用worker线程上下文(变量，函数等) 
47.   $work->stack(new Task($i)); 
48. } 
49.  
50. //循环的清理任务，会阻塞主线程，直到栈中任务都执行完毕 
51. while ($work->collect()) ; 
52.  
53. //关闭worker 
54. $work->shutdown(); 

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

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

把上面的代码修改一下：

1. <?php 
2. //创建自定义work类，给work取个名字，方便查看 
3. class Work extends Worker 
4. { 
5.   private $name; 
6.  
7.   public function __construct($name) 
8.   { 
9.     $this->name = $name; 
10.   } 
11.  
12.   public function getName() 
13.   { 
14.     return $this->name; 
15.   } 
16. } 
17.  
18. class Task extends Thread 
19. { 
20.   private $num; 
21.  
22.   public function __construct($num) 
23.   { 
24.     $this->num = $num; 
25.   } 
26.  
27.   public function run() 
28.   { 
29.     //计算累加和 
30.     $total = 0; 
31.     for ($i = 0; $i < $this->num; $i++) { 
32.       $total += $i; 
33.     } 
34.     echo "work : {$this->worker->getName()} task : {$total} \n"; 
35.     sleep(1); 
36.   } 
37. } 
38.  
39. //创建二个worker线程 
40. $work1 = new Work('a'); 
41. $work2 = new Work('b'); 
42.  
43. $work1->start(); 
44. $work2->start(); 
45.  
46. for ($i = 1; $i <= 10; $i++) { 
47.   if ($i <= 5) { 
48.     $work1->stack(new Task($i)); 
49.   } else { 
50.     $work2->stack(new Task($i)); 
51.   } 
52. } 
53.  
54. //循环的清理任务，会阻塞主线程，直到栈中任务都执行完毕 
55. while ($work1->collect() || $work2->collect()) ; 
56.  
57. //关闭worker 
58. $work1->shutdown(); 
59. $work2->shutdown(); 

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

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

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

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

1. <?php 
2. class DB extends Worker 
3. { 
4.   //注意这里设置为静态成员，pdo连接本身是不能在上下文中共享的 
5.   //声明为静态成员，让每个worker有自已的pdo连接 
6.   private static $db = null; 
7.   public $msg = 'i from db'; 
8.  
9.   public function run() 
10.   { 
11.     self::$db = new PDO('mysql:host=192.168.33.226;port=3306;dbname=test;charset=utf8', 'root', ''); 
12.   } 
13.  
14.   public function getDb() 
15.   { 
16.     return self::$db; 
17.   } 
18. } 
19.  
20. class Task extends Thread 
21. { 
22.   private $id; 
23.   //注意，这里不要给成员设置默认值，$result成员是线程对象是不可变的，不能被改写 
24.   private $result; 
25.  
26.   public function __construct($id) 
27.   { 
28.     $this->id = $id; 
29.   } 
30.  
31.   public function run() 
32.   { 
33.     //获取worker中的数据库连接 
34.     $db = $this->worker->getDb(); 
35.     $ret = $db->query("select * from tb_user where id = {$this->id}"); 
36.     $this->result = $ret->fetch(PDO::FETCH_ASSOC); 
37.     //访问worker中的成员变量msg 
38.     echo "data : {$this->result['id']} {$this->result['name']} \t worker data : {$this->worker->msg} \n"; 
39.   } 
40. } 
41.  
42. //创建一个worker线程 
43. $work = new DB(); 
44.  
45. $work->start(); 
46.  
47. for ($i = 1; $i <= 5; $i++) { 
48.   $work->stack(new Task($i)); 
49. } 
50.  
51. //循环的清理任务，会阻塞主线程，直到栈中任务都执行完毕 
52. while ($work->collect()) ; 
53.  
54. //关闭worker 
55. $work->shutdown(); 

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

运行结果如下：

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

1. <?php 
2. //之所以要创建一个Id线程类，主要是为了给work取个不同的ID，方便查看，哪些task线程属于哪个work中 
3. class Id extends Thread 
4. { 
5.   private $id; 
6.  
7.   public function getId() 
8.   { 
9.     //防止出现id混乱，这里使用同步操作 
10.     $this->synchronized(function () { 
11.       ++$this->id; 
12.     }); 
13.     return $this->id; 
14.   } 
15. } 
16.  
17. class Work extends Worker 
18. { 
19.   private $id; 
20.  
21.   public function __construct(Id $obj) 
22.   { 
23.     $this->id = $obj->getId(); 
24.   } 
25.  
26.   public function getId() 
27.   { 
28.     return $this->id; 
29.   } 
30. } 
31.  
32. class Task extends Thread 
33. { 
34.   private $num = 0; 
35.  
36.   public function __construct($num) 
37.   { 
38.     $this->num = $num; 
39.   } 
40.  
41.   //计算累加和 
42.   public function run() 
43.   { 
44.     $total = 0; 
45.     for ($i = 0; $i < $this->num; $i++) { 
46.       $total += $i; 
47.     } 
48.     echo "work id : {$this->worker->getId()} task : {$total} \n"; 
49.   } 
50. } 
51.  
52. //创建pool，可容纳3个work对象 
53. $pool = new Pool(3, 'Work', [new Id()]); 
54.  
55. //循环的把20个task线程提交到pool中的work对象上运行 
56. for ($i = 1; $i <= 20; $i++) { 
57.   $pool->submit(new Task($i)); 
58. } 
59.  
60. //循环的清理任务，会阻塞主线程，直到任务都执行完毕 
61. while ($pool->collect()) ; 
62.  
63. //关闭pool 
64. $pool->shutdown(); 

运行结果如下：

Tags: pthreads worker pool

分享到：