PHP swoole中使用task进程异步的处理耗时任务应用案例分析
本文实例讲述了PHPswoole中使用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进行累加和的计算例子
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进程
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=newswoole_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相关内容感兴趣的读者可查看本站专题:《PHP网络编程技巧总结》、《phpsocket用法总结》、《php面向对象程序设计入门教程》、《PHP数据结构与算法教程》及《php程序设计算法总结》
希望本文所述对大家PHP程序设计有所帮助。
声明:本文内容来源于网络,版权归原作者所有,内容由互联网用户自发贡献自行上传,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任。如果您发现有涉嫌版权的内容,欢迎发送邮件至:czq8825#qq.com(发邮件时,请将#更换为@)进行举报,并提供相关证据,一经查实,本站将立刻删除涉嫌侵权内容。