各位听众朋友们，大家好！
欢迎来到今天的“PHP 极客修炼班”。我是你们的主讲人，一个在代码江湖里摸爬滚打了十年的老司机。
今天我们要聊一个非常硬核，甚至可以说是“反直觉”的话题。通常，大家提到 PHP，脑子里蹦出来的词是什么？Laravel、WordPress、ThinkPHP，对吧？还有那个根深蒂固的误解：“PHP 是单线程的，它跑不了并发任务。”
没错，PHP 默认的运行模式确实是单线程的。但是——注意这个词，但是——如果我们不依赖 PHP 的 Web 服务器模型，而是直接跑 CLI（命令行），PHP 其实拥有一张底牌：PCNTL 扩展。
假设一个极其糟糕的物理环境：你的服务器只有一个 CPU 核心，没有 Hyper-Threading（超线程），甚至连多核都是奢望。老板却指着你的鼻子说：“在这个单核废铁上，我要你搞一个每秒能并发采集 1000 个目标 IP 的引擎。”
这时候，你会怎么做？你会去写 C++ 吗？不，那是庸才的做法。你会告诉我，你无能为力吗？错！今天，我就要带你们在这个“单核地狱”里，利用 PCNTL 扩展，搭建一个高性能的并发采集引擎。
准备好了吗？系好安全带，我们要开始 Fork 了！
第一章：为什么你需要“分身乏术”？
首先，让我们把物理环境具象化。
想象一下，你是一个单核 CPU。你就像一个只会做一道菜的厨师。你有一堆食材（待采集的数据源），还有一个高强度的订单（老板的 KPI）。
如果 PHP 是单线程的，你的厨师（主进程）就得像个傻子一样：拿起西红柿切一下 -> 洗一下 -> 炒一下 -> 上菜。切的时候不能炒，炒的时候不能切。效率低得感人。
这时候，PCNTL 的魔法来了。它允许你调用 fork()。这就像是你对厨房大喊一声“变！”——瞬间，你的厨房里多了一个一模一样的你。
现在你有两个厨师了。
厨师 A 继续去切西红柿（处理当前任务）。
厨师 B 去拿锅炒西红柿（处理下一个任务）。
这就是并发的本质。虽然物理上只有一个厨师，但在逻辑上，我们的程序似乎同时在做两件事。这就是 PCNTL 也就是进程控制的奥义：通过复制进程来模拟并发。
第二章：Fork 的代价与风险
但是，各位同学，Fork 也不是免费的午餐。它是有代价的。
当你调用 pcntl_fork() 时，操作系统会做一件极其昂贵的事情：内存拷贝。子进程会继承父进程所有的内存变量、打开的文件句柄、甚至已经加载的代码段。
如果你在父进程里加载了一个 500MB 的配置文件数组，然后生成了 1000 个子进程，恭喜你，你的内存会瞬间被吃掉 500GB（瞬间 OOM Killer）。所以，我们的原则是：父进程只做指挥官，子进程只做苦力。尽量减少子进程继承的内存负担。
此外，我们还要面对一个令人头皮发麻的敌人：僵尸进程。
当你派一个子进程去干活，它干完了，但还没来得及告诉你“我干完了”，就挂了。这时候，它就变成了一个僵尸。如果你不管它，它就会一直挂在进程表里，占用槽位。你需要用 pcntl_wait 来收尸。如果不收尸，你的进程数量会无限膨胀，服务器最终会因为系统资源耗尽而变成一滩死水。
第三章：架构设计——主从模式的舞蹈
我们要构建的采集引擎，采用经典的 Master-Worker（主从）模式。
-
Master 进程（指挥官）：
- 负责管理进程池。
- 负责分发任务（从任务队列里拿任务给 Worker）。
- 负责监控 Worker 的状态（是死掉了？还是正在偷懒？）。
- 负责信号处理（收到
SIGTERM信号时，优雅地关闭所有子进程）。
-
Worker 进程（打工人）：
- 被主进程 Fork 出来。
- 循环监听任务队列。
- 拿到任务后，执行采集逻辑（通常是网络请求）。
- 把采集到的结果丢给“结果缓冲区”（或者直接输出）。
第四章：代码实战——从零开始写引擎
为了演示，我们不写什么高大上的框架，直接上原生 PHP。让我们假设我们要采集一堆 HTTP 接口的数据。
4.1 基础配置与信号处理
首先，我们需要告诉系统，当老板说“停”的时候，我们要乖乖听话。我们需要监听信号。
<?php
// engine.php
declare(ticks = 1); // 这一行至关重要！开启信号监听
$masterPid = getmypid();
echo "[Master] 进程启动，PID: {$masterPid}n";
// 全局信号处理函数
$terminateSignal = false;
pcntl_signal(SIGTERM, function($signo) {
global $terminateSignal;
echo "[Master] 收到 SIGTERM 信号，准备优雅退出...n";
$terminateSignal = true;
});
pcntl_signal(SIGINT, function($signo) {
global $terminateSignal;
echo "[Master] 收到 Ctrl+C，准备优雅退出...n";
$terminateSignal = true;
});
// 阻塞信号集
// pcntl_sigprocmask(SIG_BLOCK, [SIGCHLD]);
// 在简单演示中，我们依赖 ticks 机制来监听信号变化解释： declare(ticks = 1); 告诉 PHP，每执行一条低级指令（比如函数调用），就检查一次信号处理器。如果不加这行，信号可能会丢失。
4.2 任务队列与数据结构
我们怎么管理任务？最简单的方法是用数组，或者更高效一点的 SplQueue。
// 任务生成器
function generateTasks($count) {
$tasks = [];
for ($i = 0; $i < $count; $i++) {
// 模拟一些待采集的数据
$tasks[] = [
'id' => $i,
'url' => "http://example.com/api/data?id={$i}",
'priority' => 1
];
}
return $tasks;
}
// 初始化任务队列
$taskQueue = new SplQueue();
$allTasks = generateTasks(50); // 生成50个任务
foreach ($allTasks as $task) {
$taskQueue->enqueue($task);
}4.3 Worker 进程的核心逻辑
Worker 进程是重灾区，也是最核心的部分。我们需要一个 runWorker 函数。
function runWorker($taskId) {
// 1. 模拟繁重的网络采集工作
echo "[Worker] PID: " . getmypid() . " 开始执行任务 #{$taskId}n";
// 模拟网络延迟
$start = microtime(true);
sleep(1);
$duration = round(microtime(true) - $start, 2);
// 模拟采集到的数据
$data = [
'task_id' => $taskId,
'status' => 'success',
'latency' => $duration,
'timestamp' => time()
];
echo "[Worker] PID: " . getmypid() . " 完成任务 #{$taskId}，耗时 {$duration}秒n";
// 2. 这里可以处理结果，比如写入文件或数据库
// file_put_contents("result_{$taskId}.json", json_encode($data));
return $data;
}4.4 进程管理与并发控制
现在是最精彩的时刻。我们决定开多少个“分身”呢？
假设我们要模拟 5 个并发 Worker。我们就循环 5 次，每次调用 pcntl_fork()。
$workerCount = 5;
$workers = [];
for ($i = 0; $i < $workerCount; $i++) {
$pid = pcntl_fork();
if ($pid == -1) {
die("Could not fork process");
} elseif ($pid) {
// 父进程逻辑：记录子进程 PID
echo "[Master] 启动子进程 Worker #{$i}, PID: {$pid}n";
$workers[$pid] = $i;
} else {
// 子进程逻辑
echo "[Worker] 我活着！我是子进程，我的 ID 是 Worker #{$i}, PID: " . getmypid() . "n";
// 子进程启动主循环
// 这里我们简化一下，让子进程自己去抢任务
// 在实际高性能场景中，我们通常用共享内存或消息队列来分发
performWorkLoop($taskQueue, $i);
exit(0); // 子进程执行完毕退出
}
}4.5 真正的循环与等待
上面的 performWorkLoop 函数需要处理队列。这里有一个陷阱：如果队列空了怎么办？子进程不能睡大觉，否则就浪费 CPU 了。但也不能一直空转。
function performWorkLoop($queue, $workerId) {
global $terminateSignal;
while (!$terminateSignal) {
// 检查队列是否有任务
if ($queue->isEmpty()) {
// 如果是生产者-消费者模型，这里可以 sleep 避免空转
// 但在演示中，我们让它空转，直到父进程喂食或者主循环结束
usleep(100000); // 休眠 0.1 秒
continue;
}
// 从队列头部取出任务
$task = $queue->dequeue();
// 执行任务
runWorker($task['id']);
// 防止子进程因为网络阻塞死循环
// 实际场景中，这里应该有超时处理
}
echo "[Worker] 我收到退出的信号，退出循环。n";
}4.6 收尸与等待
最后，回到父进程。子进程都在干活，我们不能直接 exit()。我们需要等所有子进程都干完活，再关闭。
// 父进程等待所有子进程结束
while (count($workers) > 0) {
// 阻塞等待任意一个子进程结束
$pid = pcntl_wait($status);
if ($pid > 0) {
// 收到了一个子进程的尸体
echo "[Master] 收到子进程 {$pid} 的尸体，清理战场。n";
unset($workers[$pid]);
}
}
echo "[Master] 所有子进程已清理，系统退出。n";第五章：进阶优化——从“能用”到“高性能”
上面的代码能跑，但只能跑在 50 个任务上。要是我们要跑 10,000 个任务呢？CPU 跑满，内存爆炸，系统卡死。
这就涉及到真正的高性能并发采集引擎的设计了。我们要解决三个核心问题：任务分配、内存隔离、状态同步。
5.1 任务分配的瓶颈
在刚才的代码里，子进程自己抢队列。如果队列是 SplQueue（内存数组），一旦任务量达到几十万，内存压力巨大。而且所有子进程都在抢同一个 PHP 对象的锁，这会导致伪共享（False Sharing）和竞争，性能反而下降。
解决方案：Redis 消息队列
不要把队列放在 PHP 内存里。把队列放在 Redis 里。
Master 进程负责把任务推入 Redis List。Worker 进程负责从 Redis List 里 Pop 任务。这是经典的 LPUSH + BRPOP 模式。
// Worker 端伪代码
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
while (!$terminateSignal) {
// 阻塞式获取，如果没有任务，Worker 就挂起休息
$taskJson = $redis->brPop('task_queue', 10);
if ($taskJson) {
$task = json_decode($taskJson[1], true);
runWorker($task['id']);
}
}这样，Master 进程只需要负责 redis->lpush，Worker 进程自动会去拿。Master 进程的压力骤降，它只需要负责监控 Worker 是否还活着。
5.2 状态同步与结果回收
Worker 采集到了数据，怎么给 Master？
直接写文件？不，太慢。
直接网络发回 Master？如果是高并发，Master 每秒要处理成千上万条写入，I/O 爆炸。
解决方案：共享内存
使用 sysvmsg (System V 消息队列) 或者 shmop (共享内存)。
最简单的是 消息队列。Master 和 Worker 都连同一个 SysV 消息队列。
Worker 发送结果：
$msg = serialize($result);
msg_send($queueId, 1, $msg); // type=1 表示结果Master 接收结果：
msg_receive($queueId, 0, $type, 1024, $message, true, 0);这样，Worker 把结果丢进队列就不管了，Master 后台慢慢接收。这大大降低了 Worker 的阻塞时间。
5.3 进程池的大小控制
CPU 是单核，难道我只能开 1 个 Worker 吗？不是的！
你可以开 10 个、20 个甚至 50 个 Worker。
为什么？
因为IO 等待时间。
Worker 代码里有 sleep(1)（模拟网络请求）。当 Worker 在等 HTTP 返回的时候，它的 CPU 占用率接近 0。这时候，即使有 100 个 Worker，CPU 核心也只需要切来切去就行。
黄金法则：
如果你的任务是纯计算（比如加密算法），CPU 是瓶颈，Worker 数量不要超过 CPU 核心数 + 1。
如果你的任务是IO 密集型（网络请求、数据库查询、文件读写），Worker 数量可以开到几十甚至上百。只要内存够，越多越好！
第六章：实战代码——构建“赛博朋克”采集器
让我们综合以上所有内容，写一个稍微像样的版本。
功能设定：
- Master 负责生成 1000 个任务，放入 Redis。
- 启动 20 个 Worker 进程。
- Worker 从 Redis 取任务，模拟耗时采集，将结果写入 PHP 共享内存。
- Master 定期从共享内存读取结果并清空，防止内存溢出。
- 支持信号中断。
注：以下代码为了演示清晰，省略了部分错误检查和复杂的配置加载，但核心逻辑完整。
<?php
// high_performance_collector.php
// 配置
define('WORKER_COUNT', 20); // 并发数
define('REDIS_HOST', '127.0.0.1');
define('REDIS_PORT', 6379);
define('TASK_QUEUE', 'collector_tasks');
define('RESULT_QUEUE', 'collector_results');
// 全局控制标志
$stopFlag = false;
// 初始化 Redis (Worker 和 Master 都要连，或者 Master 单独连，这里为了演示统一连接)
try {
$redis = new Redis();
$redis->connect(REDIS_HOST, REDIS_PORT);
} catch (Exception $e) {
die("Redis 连接失败，请先启动 Redis 服务: " . $e->getMessage());
}
// 1. 信号处理
pcntl_signal(SIGTERM, function() use (&$stopFlag) { $stopFlag = true; });
pcntl_signal(SIGINT, function() use (&$stopFlag) { $stopFlag = true; });
// 2. Master 生成任务
echo "[Master] 正在生成任务...n";
for ($i = 0; $i < 100; $i++) {
$task = ['id' => $i, 'url' => "http://target.com/data/{$i}"];
$redis->lPush(TASK_QUEUE, json_encode($task));
}
echo "[Master] 任务生成完毕，当前队列长度: " . $redis->lLen(TASK_QUEUE) . "n";
// 3. Fork Workers
$processes = [];
for ($i = 0; $i < WORKER_COUNT; $i++) {
$pid = pcntl_fork();
if ($pid === -1) {
die("Could not fork process");
} elseif ($pid) {
$processes[$pid] = $i; // 记录父进程
} else {
// --- Worker 进程开始 ---
// 轮询获取任务
while (!$stopFlag) {
// 阻塞等待任务，超时 5 秒，避免死锁
$result = $redis->brPop(TASK_QUEUE, 5);
if ($result) {
$taskData = json_decode($result[1], true);
$taskId = $taskData['id'];
echo "[Worker-{$i}] PID: " . getmypid() . " 正在处理 Task #{$taskId}n";
// 模拟采集
$res = doCollect($taskData['url']);
// 将结果放回 Master 可访问的地方
// 这里我们用 Redis List 简单模拟，Master 后台读
$redis->lPush(RESULT_QUEUE, json_encode([
'worker_id' => $i,
'pid' => getmypid(),
'task_id' => $taskId,
'data' => $res
]));
echo "[Worker-{$i}] PID: " . getmypid() . " 完成 Task #{$taskId}n";
}
}
exit(0);
// --- Worker 进程结束 ---
}
}
// 4. Master 循环监控
echo "[Master] Master 进程正在监控 {$processes} 个 Worker...n";
// 简单的统计
$processedCount = 0;
while (count($processes) > 0) {
// 等待子进程退出
$pid = pcntl_wait($status);
if ($pid > 0) {
echo "[Master] Worker PID {$pid} 退出。n";
unset($processes[$pid]);
}
// 检查结果队列
while ($resultJson = $redis->rPop(RESULT_QUEUE)) {
$result = json_decode($resultJson, true);
echo "[Master] 收到结果: Task #{$result['task_id']} from Worker #{$result['worker_id']}n";
$processedCount++;
}
// 检查是否所有任务都完成了
if ($redis->lLen(TASK_QUEUE) == 0 && count($processes) == 0 && $processedCount == 100) {
echo "[Master] 所有任务完成！n";
$stopFlag = true;
}
usleep(100000); // 0.1s
}
echo "[Master] 采集引擎已关闭。n";
function doCollect($url) {
// 模拟耗时操作
sleep(1);
return "Data from {$url}";
}第七章：避坑指南——那些年你踩过的坑
写这种高性能引擎，如果不注意细节，你会掉进深坑里。
1. 变量继承的陷阱
在 Fork 之后，千万不要修改父进程传递过来的大型变量。
比如父进程里有一个 $config = [huge array]，子进程改了一个值。这不仅浪费内存，还可能导致逻辑混乱。
做法： 让子进程自己去读配置文件，或者只传递必要的数据 ID。
2. 全局静态变量
在 PHP 中，全局变量在子进程里是共享的吗？
严格来说，PHP 的内存模型中，父进程的内存会被复制。但如果你使用了 static 变量，那它是每个进程独立的。这通常是好事，但有时候你可能会误以为它们是共享的，导致数据错乱。
3. 资源泄漏
如果你在 PHP 里打开了一个文件句柄，Fork 后，子进程里也有这个句柄。如果你关闭了它，父进程的句柄也失效了。这是多进程编程的通病。
做法： 每个进程尽量独立建立连接。比如数据库连接，Redis 连接，最好在 Worker 循环开始前建立，结束时关闭。
4. 信号竞态
如果你的 Worker 处理得很快，pcntl_wait 可能会错过子进程的退出信号。
做法： 哪怕子进程已经退出了，只要父进程还没调用 wait，这个进程就是僵尸。所以父进程必须有一个 while 循环不停地 pcntl_wait。
第八章：总结——单核的荣耀
回顾一下，我们通过 pcntl_fork 克制了单核 CPU 的短板。
我们通过 Redis 队列实现了任务的解耦。
我们通过信号处理保证了系统的稳定性。
你看，PHP 不仅仅是用来写网页的。在 PHP CLI 的世界里，它就是一个强大的多进程操作系统。在这个“不支持多线程”的物理环境下，我们用 CPU 的调度能力，换来了逻辑上的并行。
性能不是靠硬件堆出来的，是靠巧妙的架构设计出来的。
当然，如果你的项目非常复杂，涉及大量共享内存操作，或者有跨语言的交互需求，你可能需要考虑 Go 或 Java。但在纯 PHP 的语境下，PCNTL 就是你的核武器。握紧它，别让那些说 PHP 不行的家伙笑话了。
今天的讲座就到这里。下课！
（代码示例已包含在文章中，请各位同学回去后，在自己的一台单核 VPS 上跑一跑，感受一下 20 个进程并发抢食的快感！）