高并發(fā)實例分享：Swoole通過聚合請求高效實現(xiàn)業(yè)務

<?php swooleRuntime::enableCoroutine($flags = SWOOLE_HOOK_ALL);  // 最大等待次數(shù) const MAX_TIMES = 10; // 按批處理時, 每一批的最大請求暫留數(shù)量 const MAX_REQUEST = 3; // 服務端最大超時時間, 避免客戶端一直等待 const MAX_TIMEOUT = 5;  Corun(function () {     // 請求傳輸?shù)腸hannel, 原因是不要在swoole的協(xié)程環(huán)境中, 使用多個協(xié)程修改同一個全局變量     // 如果是golang, 當然是可以不定義這里的$rqChannel     // 只需要簡單的將下面的$rqQueue和$times定義為全局變量即可達到一樣的效果     // 但是最好的方式任然是是通過channel共享內(nèi)存     $rqChannel = new SwooleCoroutineChannel(MAX_REQUEST);      // 模擬創(chuàng)建訂單     $createOrder = function () use ($rqChannel) {         // 使用數(shù)組模擬請求暫留隊列         $rqQueue = [];         // 使用等待次數(shù)模擬tick效果         $times = MAX_TIMES;         while (true) {             $times--;             // 必須帶上timeout參數(shù), 否則channel是阻塞的             $rq = $rqChannel->pop(1); ????????????//?保存1個正常的請求數(shù)據(jù) ????????????if?(!empty($rq))?{ ????????????????$rqQueue[]?=?$rq; ????????????} ????????????//?請求數(shù)量未達上限或者還有等待次數(shù)時,?提前進入下一次循環(huán) ????????????if?($times?&gt;?0?&amp;&amp;?count($rqQueue)??$rq)?{ ????????????????list($data,?$chan)?=?$rq; ????????????????//?這里可以考慮后置執(zhí)行,?原因是后面可以有一些補救邏輯 ????????????????unset($rqQueue[$index]); ????????????????//?判斷$chan是否關閉? ????????????????if?($chan-&gt;errCode?===?SWOOLE_CHANNEL_CLOSED)?{ ????????????????????$data?=?null; ????????????????????continue; ????????????????} ????????????????$bool?=?$validator($data); ????????????????if?($bool)?{ ????????????????????$inserts[]?=?"({$data['user_name']},?{$data['amount']},?{$data['mobile']})"; ????????????????????$chan-&gt;push(['state'?=&gt;?1]); ????????????????}?else?{ ????????????????????$chan-&gt;push(['state'?=&gt;?0]); ????????????????} ????????????????//?unset($rqQueue[$index]); ????????????} ????????????$sql?.=?(implode(',',?$inserts)?.?';'); ????????????//?模擬創(chuàng)建訂單落庫的邏輯 ????????????echo?$sql; ????????} ????};  ????//?新手要注意這一句代碼的位置,?原因是?$server-&gt;start()?之后的代碼不會執(zhí)行 ????go($createOrder);  ????//?路由處理器 ????$orderHandler?=?function?($rq,?$res)?use?($rqChannel)?{ ????????$chan?=?new?SwooleCoroutineChannel(1); ????????//?使用timeout參數(shù)模擬超時 ????????$bool?=?$rqChannel-&gt;push([$rq-&gt;post,?$chan],?MAX_TIMEOUT); ????????if?(!$bool)?{ ????????????//?關閉$chan ????????????$chan-&gt;close(); ????????????$res-&gt;end('timeout'); ????????} ????????if?(!empty($data?=?$chan-&gt;pop()))?{ ????????????//?關閉$chan ????????????$chan-&gt;close(); ????????????//?區(qū)分成功或失敗狀態(tài)再輸出響應 ????????????if?($data['state']?===?1)?{ ????????????????$res-&gt;end(microtime()); ????????????}?else?{ ????????????????$res-&gt;end('error'); ????????????} ????????} ????};  ????$server?=?new?CoHttpServer("0.0.0.0",?9502,?false);  ????$server-&gt;handle('/order/create',?$orderHandler); ????//?當前協(xié)程容器的終點 ????$server-&gt;start(); });