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（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。