php 并发和请求次数限制的实现

在PHP中

• 如何限制并发
  • 比如100个请求同时过来了，只处理前10个请求，让另外90个等待着去处理,第102个请求等待或丢弃掉。
• 如何限制单位时间内请求次数
  • 比如一秒内100个请求过来了，只处理10个请求，其他90个请求发出429状态码或者等待1秒后在继续处理10个。当第102个请求过来时等待或丢弃掉。

限制并发

传统的PHP-FPM 似乎不支持并发限制，因为PHP-FPM 是同步阻塞的，一个请求进来，一个请求处理完，再处理下一个请求。

在 reactphp/http 中 自带的有一个中间件 LimitConcurrentRequestsMiddleware 可以限制并发请求。

$http = new React\Http\HttpServer(
    new React\Http\Middleware\LimitConcurrentRequestsMiddleware(10), // 10 concurrent buffering handlers
    $handler
);

它能做到的是100个请求过来了，处理10个请求，后面的请求会一直等待，第101个请求也在等待，不会丢掉。

如何将第101个请求，丢掉呢。

在原有的基础上，记录最大总请求数，超过了就丢掉，不失为一个办法。

需要注意的是，对于返回的流响应，怎么定义一次请求的完成，是一种业务上的选择。上方的方式仅仅是header准备好，就算一次完成，对于stream body 是不是发送完成，就没有关注了。

在当前返回 text/event-stream （chatgpt 的事件流） 是一种比较常见的用法后，定义stream全部发送完成作为一次请求的finished, 在业务上就比较有了强的需求。

reactphp/http 支持stream 响应，且能知道什么时候stream发送完成。

在laravel 中怎么使用呢？需要将laravel和reactphp结合起来。

laravel-zero 是一个精简版的命令行 laravel，将其与reactphp结合起来，就能使用laravel的便利性和reactphp 的异步特性。

安装体验

composer create-project reactphp-x/reactphp-x reactphp-x dev-master -vvv

cd reactphp-x && copy .env.example .env && composer require reactphp-x/limiter-concurrent -vvv

在 route/api.php 中添加

Route::get('/concurrent', new class {

    protected $concurrent;
    public function __construct() {
        // 100个请求同时过来了，只处理前10个请求，让另外90个等待着去处理,第102个请求等待或丢弃掉(一个请求完成：header 和body 发送完毕)。
        $this->concurrent = new \ReactphpX\Concurrent\Concurrent(10, 100, true);
    }
    public function __invoke(ServerRequestInterface $request, $next) {
       return $this->concurrent->concurrent(fn () => $next($request))->then(null, function ($error) {
        // 第101 和 102 请求会返回 503 状态码
        if ($error instanceof \OverflowException) {
            \Log::info('Server busy');
            return new Response(503, [], 'Server busy');
        }
        throw $error;
    });
    }
}, function (ServerRequestInterface $request) {
    $stream = new \React\Stream\ThroughStream();
    \React\EventLoop\Loop::get()->addTimer(1, function () use ($stream) {
        $stream->end("Hello wörld!\n");
    });
    return new Response(200, ['Content-Type' => 'text/plain; charset=utf-8'], $stream);
});

在终端运行

php artisan reactphp:http start

访问 127.0.0.1:8082/concurrent

测试

ab -n 102 -c 102 http://127.0.0.1:8082/concurrent

看看日志里是不是有 2个 Server busy？

限制请求次数

在laravel 中

Route::middleware('throttle:10,1')->group(function () {
    Route::get('/user-data', function (Request $request) {
        // 处理用户数据的逻辑
    }); 
});

这样只是返回429 状态码，可以block 1秒后再去处理，那样系统就阻塞在那里了，整体性能会下降。

请求次数的限制，可以使用令牌桶算法，每秒生成10个令牌，请求过来，拿一个令牌，没有令牌就返回429状态码或继续等待直到获取到令牌。

reactphp-x/limiter 实现了令牌桶算法。接上方的例子，只需要在route/api.php 中添加

Route::get('/limiter', new class {
    protected $limiterConcurrent;

    public function __construct() {
        // 一秒内100个请求过来了，只处理10个请求，其他90个请求发出429状态码或者等待1秒后在继续处理10个。当第102个请求过来时等待或丢弃掉 （一个请求完成：header 和body 发送完毕）。
        $this->limiterConcurrent = new \ReactphpX\LimiterConcurrent\LimiterConcurrent(10, 1000, false, 100, true);
        $this->limiterConcurrent->release(10);
    }
    public function __invoke(ServerRequestInterface $request, $next) {

        // 立即 429
        // if (!$this->limiterConcurrent->tryAcquire(1)) {
        //     return new Response(429, [], 'Too many requests');
        // }
        // return $next($request);

        // 一秒内100个请求过来了，只处理10个请求，其他90个请求等待1秒后在继续处理10个。当第102个请求过来时等待或丢弃掉 （一个请求完成：header 和body 发送完毕）
        return $this->limiterConcurrent->concurrent(fn () => $next($request))->then(null, function ($error) {
            // 第101 和 102 请求会返回 503 状态码
            if ($error instanceof \OverflowException) {
                \Log::info('limiter Server busy');
                return new Response(503, [], 'Server busy');
            }
            throw $error;
        });
    }
}, function (ServerRequestInterface $request) {
    $stream = new \React\Stream\ThroughStream();
    \React\EventLoop\Loop::get()->addTimer(1, function () use ($stream) {
        $stream->end("Hello wörld!\n");
    });
    return new Response(200, ['Content-Type' => 'text/plain; charset=utf-8'], $stream);
});

测试

ab -n 102 -c 102 http://127.0.0.1:8082/limiter

看看日志里是不是有 2个 limiter Server busy？

这些有什么用？

你的业务不需要异步操作，这些其实用处不大，但是当你的业务需要异步操作时，这些就能发挥作用了。

对于当下的gpt而言，chatgpt 会话模型，对于一个用户的请求，需要等待上一个请求完成后才能继续下一个请求，这时候就需要限制并发请求，保证每个用户能有合理的调用频率。

或者你有一个gpt和画图的token池，需要对token池进行限制，平台的token并发和请求次数都是有限制的，这时候做层限制，保证api的稳定性。

本作品采用《CC 协议》，转载必须注明作者和本文链接