本文目录导读：
在 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
核心要点：
- 任务粒度：保证单个任务执行时间 < 1秒
- 资源复用：进程池化，避免重复创建
- 错误隔离：单个任务失败不影响整体
- 监控告警：实时监控队列长度和消费速度
选择方案时，如果是小型项目用 pcntl_fork 即可，中大型项目建议使用 Supervisor + RabbitMQ/Redis Stream，性能要求极高的话使用 Swoole 的协程模型。
