PHP 怎么预留并发

wen PHP项目 2026-08-04 298

本文目录导读：

PHP 怎么预留并发

预留并发的基础认知
按架构层级预留并发
预留并发的三种经典模式
预留并发的配置清单
预留并发的高级技巧
预留并发的监控指标

在 PHP 中预留并发处理能力，主要取决于你的部署架构和业务场景，以下我按从简单到复杂、从单机到分布式的顺序，给出完整的预留方案和代码示例。

预留并发的基础认知

PHP 的并发能力受限于 PHP-FPM 进程数 和 数据库连接池，预留并发不是简单的加 Swoole，而是架构层面的设计。

按架构层级预留并发

传统 PHP-FPM 架构（最容易预留）

核心思想：进程隔离 + 队列削峰

// config/fpm_pool.conf 预留策略
pm = dynamic
pm.max_children = 100        // 预留最大并发进程数
pm.start_servers = 20        // 初始进程数
pm.min_spare_servers = 10    // 最小空闲进程（预留响应能力）
pm.max_spare_servers = 30    // 最大空闲进程
pm.max_requests = 500        // 防止内存泄漏

代码层面预留并发（使用 Redis 队列异步化）

<?php
// 1. 同步请求 → 转异步队列，预留并发处理能力
function handleHighConcurrency($requestData) {
    // 将任务推入 Redis 队列，由 Worker 进程消费
    $redis = new Redis();
    $redis->connect('127.0.0.1', 6379);
    $redis->lPush('task_queue', json_encode($requestData));
    // 立即返回给用户，不阻塞等待
    return response()->json(['status' => 'accepted']);
}
// 2. 独立的 Worker 进程（需要额外进程池）
// worker.php
while (true) {
    $task = $redis->brPop('task_queue', 0);
    processTask($task);
}

预留数据库并发（最常见的瓶颈）

方案：连接池 + 读写分离预分配

<?php
// 使用 PDO 预留连接池（单进程内复用）
class DBConnectionPool {
    private array $pool = [];
    private int $maxSize = 50; // 预留 50 个连接
    public function getConnection(): PDO {
        if (empty($this->pool)) {
            $pdo = new PDO("mysql:host=localhost;dbname=test", "user", "pass");
            $pdo->setAttribute(PDO::ATTR_PERSISTENT, true); // 持久连接
            return $pdo;
        }
        return array_pop($this->pool);
    }
    public function release(PDO $pdo): void {
        if (count($this->pool) < $this->maxSize) {
            $this->pool[] = $pdo;
        }
    }
}
// 业务代码中预留并发处理
function getProductDetail($productId) {
    try {
        $pdo = DBConnectionPool::getInstance()->getConnection();
        // 预留：将高频查询放入 Redis 缓存，减少数据库并发
        $cacheKey = "product:{$productId}";
        $redis = Redis::getInstance();
        if ($cached = $redis->get($cacheKey)) {
            return json_decode($cached, true);
        }
        // 数据库查询
        $stmt = $pdo->prepare("SELECT * FROM products WHERE id = ?");
        $stmt->execute([$productId]);
        $data = $stmt->fetch(PDO::FETCH_ASSOC);
        // 回填缓存
        $redis->setex($cacheKey, 3600, json_encode($data));
        DBConnectionPool::getInstance()->release($pdo);
        return $data;
    } catch (PDOException $e) {
        // 预留降级逻辑
        return ['error' => 'system busy'];
    }
}

高并发预留（使用 Swoole 常驻内存）

<?php
// server.php - 预留并发能力的关键
use Swoole\Http\Server;
use Swoole\Http\Request;
use Swoole\Http\Response;
// 预分配 Worker 进程数（建议 = CPU核心数 × 2）
$server = new Server("0.0.0.0", 9501);
$server->set([
    'worker_num' => 32,          // 预留 32 个 Worker 进程
    'max_request' => 10000,      // 内存清理
    'task_worker_num' => 16,     // 预留异步任务 Worker
    'backlog' => 256             // 连接队列长度预留
]);
// 预留协程并发处理
$server->on('request', function (Request $req, Response $resp) {
    // 使用协程处理，不阻塞 Worker
    go(function () use ($req, $resp) {
        // 业务逻辑
        $data = ['status' => 'ok', 'time' => time()];
        $resp->header('Content-Type', 'application/json');
        $resp->end(json_encode($data));
    });
});
// 预留 MySQL 连接池
$pool = new Swoole\Coroutine\Channel(100); // 预留 100 个协程连接
for ($i = 0; $i < 100; $i++) {
    $mysql = new Swoole\Coroutine\MySQL();
    $mysql->connect([
        'host' => '127.0.0.1',
        'user' => 'root',
        'password' => 'pass',
        'database' => 'test',
    ]);
    $pool->push($mysql);
}
$server->start();

全链路限流预留（防止雪崩）

<?php
// 1. 令牌桶算法 - 预留稳定的处理能力
class TokenBucket {
    private int $capacity;      // 桶容量（最大预留值）
    private float $tokens;      // 当前令牌数
    private float $refillRate;  // 补充速率（每秒）
    private float $lastRefill;  // 上次补充时间
    public function __construct(int $capacity = 1000, float $refillRate = 100) {
        $this->capacity = $capacity;
        $this->tokens = $capacity;
        $this->refillRate = $refillRate;
        $this->lastRefill = microtime(true);
    }
    public function tryAcquire(): bool {
        $now = microtime(true);
        $this->tokens = min($this->capacity, 
            $this->tokens + ($now - $this->lastRefill) * $this->refillRate);
        $this->lastRefill = $now;
        if ($this->tokens >= 1) {
            $this->tokens -= 1;
            return true;
        }
        return false;
    }
}
// 2. 使用信号量预留（进程内并发控制）
$semaphore = sem_get(12345, 10); // 预留 10 个并发槽位
if (sem_acquire($semaphore)) {
    try {
        // 临界区业务
        processRequest();
    } finally {
        sem_release($semaphore); // 释放槽位
    }
}

预留并发的三种经典模式

消峰填谷（异步队列）

用户请求 → 快速写入消息队列 → 立即返回成功 → 后台 Worker 慢速处理

// 预留实现
$mq = new RabbitMQ();
$mq->declareQueue('order_queue');
$mq->publish(json_encode($orderData)); // 1ms 完成
// 立即返回，真正处理由 10 个 Worker 消费

流量控制（滑动窗口）

class SlidingWindowLimiter {
    private Redis $redis;
    private string $key;
    private int $windowSize;     // 窗口大小
    private int $maxRequests;    // 允许最大请求数
    public function allow(): bool {
        $now = microtime(true);
        $key = "rate_limit:{$this->key}:window";
        $this->redis->multi()
            ->zRemRangeByScore($key, 0, $now - $this->windowSize)
            ->zAdd($key, $now, uniqid())
            ->zCard($key)
            ->expire($key, $this->windowSize + 1);
        $result = $this->redis->exec();
        return $result[2] <= $this->maxRequests;
    }
}

主备降级（超时中断）

// 设置超时时间，预留快速失败能力
set_time_limit(3); // 3秒超时
try {
    $response = file_get_contents('http://external-api.com/data', false, 
        stream_context_create(['http' => ['timeout' => 2]]));
} catch (Exception $e) {
    // 预留降级数据
    $response = getFromCacheBackup();
}

预留并发的配置清单

PHP-FPM 配置（php.ini）

; 预留项目
memory_limit = 256M          ; 数据量预留
max_execution_time = 30      ; 超时预留
max_input_vars = 3000        ; 参数预留
upload_max_filesize = 20M
post_max_size = 25M

Nginx 代理配置

# 预留连接数
worker_connections 1024;
# 预留超时设置
fastcgi_connect_timeout 30;
fastcgi_send_timeout 300;
fastcgi_read_timeout 300;
# 预留缓冲
fastcgi_buffers 8 16k;
fastcgi_buffer_size 32k;

数据库配置预留

-- 预留连接
SET GLOBAL max_connections = 500;
SET GLOBAL wait_timeout = 60;
SET GLOBAL interactive_timeout = 60;
-- 预留复杂查询
SET SESSION sort_buffer_size = 2M;
SET SESSION join_buffer_size = 2M;

预留并发的高级技巧

预生成数据缓冲

// 提前将热点数据放入内存
$hotData = apcu_fetch('hot_data');
if (false === $hotData) {
    $hotData = databaseCalculateHotData();
    apcu_store('hot_data', $hotData, 60); // 缓存 60 秒
}
// 直接使用，无需查询数据库

使用 Redis 分布式锁预留资源

$lock = new RedisLock($redis, 'user:123:update');
if ($lock->acquire(3)) { // 预留 3 秒锁
    try {
        // 业务处理
    } finally {
        $lock->release();
    }
}

优雅降级（Circuit Breaker）

class CircuitBreaker {
    public function call(callable $func) {
        $circuit = apcu_fetch('circuit:key');
        if ($circuit == 'open') {
            return fallback(); // 快速失败
        }
        try {
            $result = $func();
            apcu_store('circuit:key', 'closed', 10);
            return $result;
        } catch (Exception $e) {
            apcu_store('circuit:key', 'open', 5); // 熔断 5 秒
            return fallback();
        }
    }
}

预留并发的监控指标

预留后必须监控这些指标来调整：

# 查看 FPM 进程
php-fpm -tt
netstat -an | grep :9000 | grep ESTABLISHED | wc -l
# 查看队列积压
redis-cli llen task_queue
# 查看连接池
mysqladmin -u root -p status | grep Threads
# PHP 错误日志
tail -f /var/log/php-fpm.log

预留并发的核心原则：