Swoole高并发聚合请求实例

33次阅读

共计 2343 个字符，预计需要花费 6 分钟才能阅读完成。

本文旨在阐明在高并发场景下如何通过聚合申请，充分利用数据库的批量解决更高效地实现业务性能。
当然，此示例仅用作抛砖引玉，心愿能激发读者更深刻的思考。

本示例选取的背景是并发下单业务。惯例状况下，后端创立订单是逐条 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['user_name']}, {$data['amount']}, {$data['mobile']})";
                    $chan->push(['state' => 1]);
                } else {$chan->push(['state' => 0]);
                }
                // unset($rqQueue[$index]);
            }
            $sql .= (implode(',', $inserts) . ';');
            // 模仿创立订单落库的逻辑
            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('timeout');
        }
        if (!empty($data = $chan->pop())) {
            // 敞开 $chan
            $chan->close();
            // 辨别胜利或失败状态再输入响应
            if ($data['state'] === 1) {$res->end(microtime());
            } else {$res->end('error');
            }
        }
    };

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

    $server->handle('/order/create', $orderHandler);
    // 以后协程容器的起点
    $server->start();});