PHP swoole中使用task进程异步的处理耗时任务应用案例分析

本文实例讲述了PHP swoole中使用task进程异步的处理耗时任务。分享给大家供大家参考，具体如下：

我们知道，swoole中有两大进程，分别是 master 主进程和 manager 管理进程。

其中 master 主进程中会有一个主 reactor 线程和多个 reactor 线程，主要的作用就是用来维护TCP连接，处理网络IO，收发数据。

而 manager 管理进程，作用则是 fork 和管理 worker 和 task 进程。

worker 进程的作用是接收 reactor 线程传递的数据，并处理数据，返回处理结果给 reactor 线程。

task 进程的作用是处理一些相对耗时的任务，task 与 worker 进程是独立的，不会影响 worker 进程处理客户端的请求。

一、task 进程的应用场景：

1、相对耗时的邮件群发，比如某某活动，需要给100W用户发送活动邮件。

2、推送某些大V的动态，比如某大V发了条新消息，粉丝需要及时获取到该动态。

二、worker 与 task 的相互关系：

1、worker 进程中能过调用 task() 来投递任务，task 进程中 通过 onTask 事件来响应投递来的任务。

2、task 进程中 通过 直接返回 或 调用 finish() 来告诉 worker 进程任务处理完毕，worker 进程中 通过 onFinish 事件响应任务完成。

三、使用 task 的前题：

1、在 Server 中 配置 task_worker_num 数量。

2、设置 Server 的 onTask 和 onFinish 事件回调函数。

四、简单的使用task进行累加和的计算例子

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46

<?php $server = new swoole_server( '0.0.0.0' , 6666);   $server ->set([    'worker_num' => 2,    'task_worker_num' => 16, ]);   $server ->on( 'WorkerStart' , function ( $server , $worker_id ) {    //注意这里，我们通过taskworker来判断是task进程还是worker进程    //需要在worker进程中调用task()，不然会报出警告    //这里会执行两遍，因为我们设置了worker_num数为2    if (! $server ->taskworker) {      echo '投递任务开始...' , PHP_EOL;      //投递32个累加计算任务给16个task进程      for ( $ix = 0; $ix < 32; $ix ++) {        //注意这里的投递是异步的        $server ->task([mt_rand(1, 100), mt_rand(1000, 9999)]);      }      echo '投递任务结束...' , PHP_EOL;    } });   //server服务必须要有onReceive回调 $server ->on( 'Receive' , function ( $server , $fd , $reactor_id , $data ) {   });   //注意，task进程完全是同步阻塞模式的 $server ->on( 'Task' , function ( $server , $task_id , $src_worker_id , $data ) {    echo "task {$task_id} 进程正在工作..." , PHP_EOL;    $start = $data [0];    $end = $data [1];    $total = 0;    for (; $start <= $end ; $start ++) {      $total += $start ;    }    echo "task {$task_id} 进程完成工作..." , PHP_EOL;    return $total ; });   $server ->on( 'Finish' , function ( $server , $task_id , $data ) {    echo "task {$task_id} 进程处理完成, 结果为 {$data}" , PHP_EOL; });   $server ->start();

注意，我们通过调用 task() 往任务池中投递任务，swoole 底层会轮询的投递任务到各个 task 进程。

当你投递任务的数量超过 onTask 的处理速度，这会导致任务池被塞满，进而导致 worker 进程发生阻塞，所以需合理设置 task_worker_num 数量和处理速度之间的关系。

当然，我们也可以人为的把任务投递到指定的 task 进程。task() 函数的第二个参数可以指定要投递的 task 进程ID，ID范围为 0 到 (task_worker_num - 1)。

五、对任务进行切分，人为控制投递到 task 进程

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56

<?php $server = new swoole_server( '0.0.0.0' , 6666);   $server ->set([    'worker_num' => 1,    'task_worker_num' => 10, ]);   $server ->on( 'WorkerStart' , function ( $server , $worker_id ) {    //为了方便演示，把worker_num设置为1，这里只会执行一次    if (! $server ->taskworker) {      //通过swoole_table共享内存，在不同进程中共享数据      $server ->result = new swoole_table(10240);      //用于保存task进程完成数量      $server ->result->column( 'finish_nums' , swoole_table::TYPE_INT);      //用于保存最终计算结果      $server ->result->column( 'result' , swoole_table::TYPE_INT);      $server ->result->create();      //计算1000的累加和，并把计算任务分配到10个task进程上      $num = 1000;      $step = $num / $server ->setting[ 'task_worker_num' ];      for ( $ix = 0; $ix < $server ->setting[ 'task_worker_num' ]; $ix ++) {        $start = $ix * $step ;        $server ->task([ $start , $start + $step ], $ix );      }    } });   $server ->on( 'Receive' , function ( $server , $fd , $reactor_id , $data ) {   });   //注意，task进程完全是同步阻塞模式的 $server ->on( 'Task' , function ( $server , $task_id , $src_worker_id , $data ) {    echo "task {$task_id} 进程正在工作... 计算 {$data[0]} - {$data[1]} " , PHP_EOL;    $start = ++ $data [0];    $end = $data [1];    $total = 0;    for (; $start <= $end ; $start ++) {      $total += $start ;    }    echo "task {$task_id} 进程完成工作..." , PHP_EOL;    return $total ; });   $server ->on( 'Finish' , function ( $server , $task_id , $data ) {    echo "task {$task_id} 进程处理完成, 结果为 {$data}" , PHP_EOL;    $server ->result->incr( 'finish_nums' , 'finish_nums' );    $server ->result->set( 'result' , [ 'result' => $data + $server ->result->get( 'result' , 'result' )]);      if ( $server ->result->get( 'finish_nums' , 'finish_nums' ) == $server ->setting[ 'task_worker_num' ]) {      echo "最终计算结果：{$server->result->get('result', 'result')}" , PHP_EOL;    } });   $server ->start();

希望本文所述对大家PHP程序设计有所帮助。

原文链接：https://www.cnblogs.com/jkko123/p/10883066.html