PHP 怎么设置消费并发

wen PHP项目 2026-08-05 234

本文目录导读：

PHP 怎么设置消费并发

基础方案（进程级别）
生产级方案（推荐）
进阶配置（高并发调优）
性能监控与优化
最佳实践建议
快速上手指南

在 PHP 中实现消费并发（如消息队列消费者、任务处理）有多种方案，这里给你一个完整的指南,从简单到复杂：

基础方案（进程级别）

多进程并发（pcntl_fork）

<?php
$processCount = 4; // 并发数
$queue = new RedisQueue(); // 假设你的队列
for ($i = 0; $i < $processCount; $i++) {
    $pid = pcntl_fork();
    if ($pid == -1) {
        die("无法创建子进程");
    } elseif ($pid) {
        // 父进程继续创建
        continue;
    } else {
        // 子进程消费
        while (true) {
            $task = $queue->pop();
            if ($task) {
                handleTask($task);
            } else {
                usleep(500000); // 无任务时休眠
            }
        }
        exit;
    }
}
?>

多进程 + 信号控制

<?php
class Consumer {
    private $workers = [];
    public function start($count) {
        for ($i = 0; $i < $count; $i++) {
            $pid = pcntl_fork();
            if ($pid == -1) {
                continue;
            } elseif ($pid) {
                $this->workers[] = $pid;
            } else {
                $this->worker($i);
                exit;
            }
        }
        // 父进程等待子进程
        while ($this->workers) {
            $pid = pcntl_waitpid(-1, $status);
            $key = array_search($pid, $this->workers);
            unset($this->workers[$key]);
        }
    }
    private function worker($id) {
        echo "Worker $id 启动\n";
        while (true) {
            // 消费逻辑
            usleep(100);
        }
    }
}
$consumer = new Consumer();
$consumer->start(4);
?>

生产级方案（推荐）

使用 Supervisor 管理多进程

supervisord.conf:

[program:worker]
process_name=%(program_name)s_%(process_num)02d
command=php /path/to/worker.php
numprocs=10               ; 并发进程数
directory=/path/to/project
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=/var/log/worker.log
stopwaitsecs=2

worker.php:

<?php
require_once 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
class Worker {
    private $connection;
    private $channel;
    public function __construct() {
        $this->connection = new AMQPStreamConnection(
            'localhost', 5672, 'guest', 'guest'
        );
        $this->channel = $this->connection->channel();
        $this->channel->queue_declare('task_queue', 
            false, true, false, false
        );
        // 公平分发：每次只取一个
        $this->channel->basic_qos(
            null, 1, null
        );
    }
    public function run() {
        echo "Worker 启动\n";
        $callback = function($msg) {
            echo "收到: ", $msg->body, "\n";
            $result = $this->process($msg->body);
            if ($result) {
                $msg->delivery_info['channel']->basic_ack(
                    $msg->delivery_info['delivery_tag']
                );
            }
        };
        $this->channel->basic_consume(
            'task_queue', '', false, false, false, false, $callback
        );
        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }
    }
    private function process($data) {
        // 业务处理逻辑
        return true;
    }
}
$worker = new Worker();
$worker->run();
?>

进阶配置（高并发调优）

设置进程数的最佳实践

<?php
// 根据 CPU 核数动态设置并发
$cpuCount = shell_exec("nproc") ?: 4;
$workerCount = min(20, $cpuCount * 2);
// 创建 worker 池
class WorkerPool {
    private $workers = [];
    private $maxWorkers;
    public function __construct($maxWorkers) {
        $this->maxWorkers = $maxWorkers;
    }
    public function run($taskData) {
        while (count($this->workers) >= $this->maxWorkers) {
            // 检查worker完成情况
            usleep(100);
        }
        $pid = pcntl_fork();
        if ($pid > 0) {
            $this->workers[] = $pid;
        } else {
            try {
                doTask($taskData);
            } finally {
                exit(0);
            }
        }
    }
    public function waitAll() {
        foreach ($this->workers as $pid) {
            pcntl_waitpid($pid, $status);
        }
    }
}
?>

使用消息队列高级特性

<?php
// Redis Stream 消费者组
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$group = 'group1';
$consumer = 'consumer-' . getmypid();
// 创建消费者组
$redis->xGroup('CREATE', 'task_stream', $group, 0, true);
// 消费消息
while (true) {
    $messages = $redis->xReadGroup(
        $group,
        $consumer,
        ['task_stream' => '>'],
        1,  // 每次读取数量
        1000  // 阻塞时间ms
    );
    if ($messages) {
        foreach ($messages['task_stream'] as $id => $data) {
            // 处理消息
            handleTask($data);
            // 确认消息
            $redis->xAck('task_stream', $group, [$id]);
        }
    } else {
        usleep(500000);
    }
}
?>

性能监控与优化

任务超时控制

<?php
class TaskExecutor {
    private $timeout = 30; // 秒
    public function execute($task) {
        $pid = pcntl_fork();
        if ($pid > 0) {
            // 父进程等待，带超时
            $time = 0;
            while ($time < $this->timeout) {
                $res = pcntl_waitpid($pid, $status, WNOHANG);
                if ($res > 0) {
                    return true; // 任务完成
                }
                sleep(1);
                $time++;
            }
            // 超时杀掉子进程
            posix_kill($pid, SIGKILL);
            throw new Exception("任务超时");
        } else {
            // 子进程执行任务
            try {
                processTask($task);
                exit(0);
            } catch (Exception $e) {
                exit(1);
            }
        }
    }
}
?>

错误重试机制

<?php
class RetryHandler {
    private $maxRetries = 3;
    private $retryDelays = [1, 5, 15]; // 秒
    public function retry($task, $func) {
        for ($attempt = 0; $attempt < $this->maxRetries; $attempt++) {
            try {
                $result = $func($task);
                return $result;
            } catch (Exception $e) {
                echo "第" . ($attempt + 1) . "次失败: " . $e->getMessage() . "\n";
                if ($attempt < $this->maxRetries - 1) {
                    sleep($this->retryDelays[$attempt]);
                }
            }
        }
        throw new Exception("任务处理失败");
    }
}
?>

最佳实践建议

部署架构

┌───────────────────────────────┐
│  Supervisor / K8s             │
│  ┌───────┐ ┌───────┐ ┌──────┐ │
│  │Worker1│ │Worker2│ │Worker3││ │
│  └───┬───┘ └───┬───┘ └──┬───┘ │
│      └────┬─────┴────┬───┘     │
│           ▼          ▼          │
│      ┌─────────┐ ┌─────────┐   │
│      │  Redis  │ │ DB/MySQL│   │
│      └─────────┘ └─────────┘   │
└───────────────────────────────┘

关键参数配置表

参数	建议值	说明
并发进程数	CPU核数 × 2~3	根据任务类型调整
内存限制	256M~1G	防止内存泄漏
执行超时	30~60秒	防止任务挂死
重试次数	3次	过多会堆积
队列长度	1000~10000	防止内存溢出

注意事项

<?php
// 1. 使用对象池复用连接
class DbConnectionPool {
    private static $connections = [];
    public static function getConnect() {
        if (empty(self::$connections)) {
            self::$connections[] = new PDO(...);
        }
        return array_shift(self::$connections);
    }
    public static function release($conn) {
        self::$connections[] = $conn;
    }
}
// 2. 优雅退出
pcntl_signal(SIGTERM, function() {
    // 记录当前进度
    // 关闭资源
    exit(0);
});
// 3. 内存监控
while (true) {
    $memory = memory_get_usage(true);
    if ($memory > 256 * 1024 * 1024) {
        echo "内存超限，退出\n";
        exit;
    }
    // 处理任务
}
?>

快速上手指南

# 1. 安装依赖
composer require swoole/redis
# 2. 启动 Worker
php worker.php start --concurrency=10
# 3. 监控状态
php worker.php status
# 4. 平滑重启
php worker.php reload

核心要点：