swoole(6)Task异步任务

一:什么是task进程?

task进程是独立与worker进程的一组进程 ,他主要处理耗时较长的业务逻辑,并且不影响worker进程处理客户端的请求.worker进程通过task()函数把数据投递到Task进程去处理

开启task功能(task功能是默认关闭的 ,开启task功能需要满足2个条件)

配置task进程的数量(task_worker_num)
注册task的回掉函数onTask和onfinish

二:task设计流程

worker进程中,我们调用对应的task()方法发送数据通知到task worker进程
task worker进程会在onTask()回调中接收到这些数据,并进行处理
处理完会可以通过finish函数或直接return消息给worker进程
worker进程在onFinish()进程收到消息并进行处理

代码:

<?php

/**

 * Created by PhpStorm.

 * User: huahua

 * Date: 2020/3/6

 * Time: 上午10:09

 */

//基于第一个 TCP 服务器，只需要增加 onTask 和 onFinish2 个事件回调函数即可。

//另外需要设置 task 进程数量，可以根据任务的耗时和任务量配置适量的 task 进程。

$serv = new \Swoole\Server('127.0.0.1',9800);

$serv->set([

    'worker_num'=>2, //设置进程

    //配置此参数后将会启用 task 功能。所以 Server 务必要注册 onTask、onFinish 2 个事件回调函数。如果没有注册，服务器程序将无法启动。

    'task_worker_num' => 4, //配置 Task 进程的数量。

    'task_ipc_mode'=>2,//设置 Task 进程与 Worker 进程之间通信的方式

]);

//监听连接进入事件

$serv->on('Connect', function ($serv, $fd) {

    echo "Client: Connect.\n";

});

//监听数据接收事件

$serv->on('Receive', function ($serv, $fd, $from_id, $data) {

    $data=['tid'=>time()];

    $task_id = $serv->task($data);////投递到taskWorker进程组

    echo "task id = {$task_id},from id = $from_id".PHP_EOL;

});

$serv->on('task',function ($serv, $task_id, $from_id, $data){

    echo "task id = {$task_id},from id = $from_id,进程id =".posix_getpid().PHP_EOL;

    echo '我是task任务';

    $serv->finish("data -> ok");

});

$serv->on('finish',function ($serv, $task_id, $data){

    echo "task id = {$task_id}".PHP_EOL;

});

//监听连接关闭事件

$serv->on('Close', function ($serv, $fd) {

    echo "Client: Close.\n";

});

//启动服务器

$serv->start();

cli下运行结果:

$task_id不等于workerStart中的worker_id,而是swoole维护的任务自增长id

$from_id表示来自于哪个woker投递而来的任务，id等于worker_id值

$serv->worker_id等于workerStart中的worker_id, 此处是task_woker的id

$serv->worker_pid 此处是task_woker的进程pid

三:task任务切分

场景：假设有一台服务器专门处理前台投递的数据,利用简单的任务拆分,分配到相应的进程去处理

思路:

1.将一个大的任务拆分成相应份数(是由$task_worker_num数量来确定)
2.通过foreach循环将数据投递到指定的task进程,范围是(0-(task_worker_num-1))区间之内
3.执行失败的任务，需要保留，重新投递执行（进程间通讯管道方式）

$server->on('receive',function (swoole_server $server, int $fd, int $reactor_id, string $data){

    for ($i=0;$i<100;$i++){

        $tasks[] =['id'=>$i,'msg'=>time()];

    }

    $count=count($tasks);

    $data=array_chunk($tasks,ceil($count/3));

    foreach ($data as $k=>$v){

        $server->task($v,$k);  //(0-task_woker_num-1)

    }

});

巴特西

swoole(6)Task异步任务

一:什么是task进程?

二:task设计流程

三:task任务切分

最新文章

热门文章