本篇文章介紹了swoole中使用task進程來處理耗時任務的方法，具有一定的參考價值，希望對正在學習swoole框架的同學有幫助！
如何在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發了條新消息，粉絲需要及時獲取到該動態。
推薦學習：?swoole教程
二、worker 與 task 的相互關系：
1、worker 進程中能過調用 task() 來投遞任務，task 進程中 通過 onTask 事件來響應投遞來的任務。
2、task 進程中 通過 直接返回 或 調用?finish() 來告訴 worker 進程任務處理完畢，worker 進程中 通過?onFinish 事件響應任務完成。
三、使用 task 的前題：
1、在 Server 中 配置?task_worker_num 數量。
2、設置 Server 的?onTask 和 onFinish 事件回調函數。
四、簡單的使用task進行累加和的計算例子
<?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?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?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 進程
<?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?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?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->s tart();
