高並發實例分享：Swoole透過聚合請求高效實現業務

本篇文章跟大家分享一個Swoole高並發聚合請求實例，介紹在高並發場景下如何透過聚合請求，充分利用資料庫的批次處理更有效率地實現業務功能。此範例僅用作拋磚引玉，希望能激發大家更深入的思考。

相關影片課程推薦：《千萬級資料並發解決方案（理論實戰）》

分享一些高並發面試題：15個PHP關於高並發的面試題（總結）

本範例選取的背景是並發下單業務。在常規情況下，後端建立訂單是逐條 insert 的操作。在並發較低的時候，資料庫的insert 操作的確能保持不錯的效率，但是當遇到請求數量增多，資料庫頻繁地單次insert就會讓下單業務整體效率變低（本文簡單假設1次下單=1個insert）。

透過上面的描述，其實已經很容易想到需要優化的地方了。類比現實生活中搭乘電梯的場景：一架電梯 裝滿後再上行，可以最快緩解人流壓力。

下面我們就來用程式碼簡單實作一下我們思路：

<?php

Swoole\Runtime::enableCoroutine($flags = SWOOLE_HOOK_ALL);

// 最大等待次数
const MAX_TIMES = 10;
// 按批处理时, 每一批的最大请求暂留数量
const MAX_REQUEST = 3;
// 服务端最大超时时间, 避免客户端一直等待
const MAX_TIMEOUT = 5;

Co\run(function () {
    // 请求传输的channel, 原因是不要在swoole的协程环境中, 使用多个协程修改同一个全局变量
    // 如果是golang, 当然是可以不定义这里的$rqChannel
    // 只需要简单的将下面的$rqQueue和$times定义为全局变量即可达到一样的效果
    // 但是最好的方式任然是是通过channel共享内存
    $rqChannel = new Swoole\Coroutine\Channel(MAX_REQUEST);

    // 模拟创建订单
    $createOrder = function () use ($rqChannel) {
        // 使用数组模拟请求暂留队列
        $rqQueue = [];
        // 使用等待次数模拟tick效果
        $times = MAX_TIMES;
        while (true) {
            $times--;
            // 必须带上timeout参数, 否则channel是阻塞的
            $rq = $rqChannel->pop(1);
            // 保存1个正常的请求数据
            if (!empty($rq)) {
                $rqQueue[] = $rq;
            }
            // 请求数量未达上限或者还有等待次数时, 提前进入下一次循环
            if ($times > 0 && count($rqQueue) < MAX_REQUEST) {
                continue;
            }
            // 重置等待次数
            $times = MAX_TIMES;
            // 初始化SQL
            $sql = "INSERT INTO orders VALUES ";
            $inserts = [];
            // 模拟数据验证
            $validator = function ($input): bool {
                // 为了缩减代码, 没有真的做数据验证的处理
                array_filter($input);
                return true;
            };
            // $rqQueue在协程上下文是并发安全的, 所以遍历时不用担心
            foreach ($rqQueue as $index => $rq) {
                list($data, $chan) = $rq;
                // 这里可以考虑后置执行, 原因是后面可以有一些补救逻辑
                unset($rqQueue[$index]);
                // 判断$chan是否关闭å
                if ($chan->errCode === SWOOLE_CHANNEL_CLOSED) {
                    $data = null;
                    continue;
                }
                $bool = $validator($data);
                if ($bool) {
                    $inserts[] = "({$data[&#39;user_name&#39;]}, {$data[&#39;amount&#39;]}, {$data[&#39;mobile&#39;]})";
                    $chan->push([&#39;state&#39; => 1]);
                } else {
                    $chan->push([&#39;state&#39; => 0]);
                }
                // unset($rqQueue[$index]);
            }
            $sql .= (implode(&#39;,&#39;, $inserts) . &#39;;&#39;);
            // 模拟创建订单落库的逻辑
            echo $sql;
        }
    };

    // 新手要注意这一句代码的位置, 原因是 $server->start() 之后的代码不会执行
    go($createOrder);

    // 路由处理器
    $orderHandler = function ($rq, $res) use ($rqChannel) {
        $chan = new Swoole\Coroutine\Channel(1);
        // 使用timeout参数模拟超时
        $bool = $rqChannel->push([$rq->post, $chan], MAX_TIMEOUT);
        if (!$bool) {
            // 关闭$chan
            $chan->close();
            $res->end(&#39;timeout&#39;);
        }
        if (!empty($data = $chan->pop())) {
            // 关闭$chan
            $chan->close();
            // 区分成功或失败状态再输出响应
            if ($data[&#39;state&#39;] === 1) {
                $res->end(microtime());
            } else {
                $res->end(&#39;error&#39;);
            }
        }
    };

    $server = new Co\Http\Server("0.0.0.0", 9502, false);

    $server->handle(&#39;/order/create&#39;, $orderHandler);
    // 当前协程容器的终点
    $server->start();
});

程式碼整體上還是很容易理解的，變數$rqQueue 就是類比電梯，暫留請求等待一定時間的次數$times 就是類比電梯需要等待人流依序進入。當然最在希望讀者註意的一點是：在協程環境下，不要使用共享記憶體而通信，應該使用通信來共享記憶體。

推薦學習： swoole教學

#

以上是高並發實例分享：Swoole透過聚合請求高效實現業務的詳細內容。更多資訊請關注PHP中文網其他相關文章！