PHP TrueAsync 研究笔记

概述

PHP TrueAsync 是一个革命性的项目，它将原生异步编程能力直接带入 PHP 核心。0.6.0 版本是实验性里程碑，目标让开发者可以编写真正的异步代码。

---

核心概念

1. 完全异步化的 PHP Core

这是 0.6.0 最激进的变化。传统 PHP 以同步阻塞为主，而 TrueAsync 使核心 I/O 操作真正并发：

操作类型	传统 PHP	TrueAsync
文件 I/O	阻塞	非阻塞（Linux: IO_URING）
Socket	阻塞	非阻塞
Pipe	阻塞	非阻塞
CURL	阻塞	非阻塞（比 curl_multi_exec 快 2 倍）
PDO/MySQL	阻塞	非阻塞 + 连接池
sleep()	阻塞	非阻塞

关键洞察：普通 PHP 函数本身就可以在协程中以异步方式运行，无需额外包装器。

---

2. 异步编程 API

// 核心原语
- Coroutines: spawn() 启动异步任务
- Future: 异步结果的 Promise 变体
- Awaiting: await, await_all(), await_first_success(), delay(), suspend()
- Channels: 协程间数据传递队列
- Cancellation: cancel(), protect(), timeout()
- Scope: 管理协程生命周期
- TaskGroup/TaskSet: 结构化并发
- Context: 协程上下文数据
- Pool/PDO Pool: 资源池与数据库连接池

---

3. 进程池实现模式（基于原文）

#### 3.1 基础模式：spawn + proc_open

function run_worker(string $script, array $tasks): void {
    $process = proc_open([PHP_BINARY, $script], [...], $pipes);
    foreach ($tasks as $task) {
        fwrite($pipes[0], json_encode($task) . "\n");
        $line = fgets($pipes[1]); // ← 在 TrueAsync 中是非阻塞的
        $result = json_decode(trim($line), true);
    }
    // ...
}

// 启动 10 个并行 worker
for ($i = 0; $i < 10; $i++) {
    spawn(run_worker(...), 'worker.php', $tasks);
}

关键点：

只写了一行异步代码 spawn()
proc_open, fgets, fclose 自动变为非阻塞
协程竞争 CPU 时间，但代码仍保持顺序逻辑

#### 3.2 Channel 模式：任务分发

$taskQueue = new Async\Channel(10);

// 启动 workers
for ($i = 0; $i < 10; $i++) {
    spawn(run_worker(...), 'worker.php', $taskQueue);
}

// 投喂任务
foreach ($tasks as $task) {
    $channel->send($task); // 队列满时自动挂起（背压）
}
$channel->close();

背压机制：当所有 worker 忙碌时，send() 会挂起协程，防止内存被无法处理的任务填满。

#### 3.3 TaskGroup 模式：结构化并发

$taskQueue = new Async\Channel(10);
$group = new Async\TaskGroup();

try {
    for ($i = 0; $i < 10; $i++) {
        $group->spawn(run_worker(...), 'worker.php', $taskQueue);
    }
    foreach ($tasks as $task) {
        $taskQueue->send($task);
    }
} finally {
    $taskQueue->close();
    $group->seal();           // 防止添加新协程
    $group->all()->await();   // 等待所有 worker 完成
}

取消令牌机制：

// 如果所有 worker 死亡，停止投喂任务
$taskQueue->send($task, $group->all());

#### 3.4 TaskSet + Supervisor 模式：自愈进程池

$taskQueue = new Channel(POOL_SIZE);
$group = new TaskSet();  // TaskSet 的 join* 方法会移除已完成任务

// 启动 workers
for ($i = 0; $i < POOL_SIZE; $i++) {
    $group->spawn(run_worker(...), __DIR__ . '/worker.php', $taskQueue);
}

// Supervisor：进程崩溃时自动重启
$supervisor = spawn(function () use ($group, $taskQueue): void {
    $cooldown = 5;
    $threshold = 2;
    $restarts = 0;
    $lastFail = time();
    
    while (true) {
        try {
            $group->joinNext()->await();  // 等待至少一个任务完成
        } catch (\Exception $e) {
            echo "[supervisor] Exception: {$e->getMessage()}\n";
            $group->seal();
            $taskQueue->close();
            return;
        }
        
        if ($group->isSealed() || $taskQueue->isClosed()) {
            return;
        }
        
        // 防无限重启
        $now = time();
        if (($now - $lastFail) < $cooldown) {
            if ($restarts >= $threshold) {
                echo "[supervisor] Too many failures, shutting down\n";
                $group->seal();
                $taskQueue->close();
                return;
            }
            $restarts++;
        }
        $lastFail = $now;
        
        // 重启 worker
        $group->spawn(run_worker(...), __DIR__ . '/worker.php', $taskQueue);
    }
});

设计优点：

Supervisor 对任务如何执行一无所知
执行任务的协程对 Supervisor 一无所知
低耦合，易于维护

---

4. 错误处理增强

stream_set_timeout($pipes[1], 5);  // 5 秒超时

foreach ($taskQueue as $task) {
    $encoded = json_encode($task);
    if ($encoded === false) {
        echo "[worker] json_encode failed\n";
        return;
    }
    
    if (!fwrite($pipes[0], $encoded . "\n")) {
        echo "[worker] pipe broken\n";
        return;
    }
    
    $line = fgets($pipes[1]);
    if ($line === false && stream_get_meta_data($pipes[1])['timed_out']) {
        echo "[worker] timeout\n";
        return;
    } elseif ($line === false) {
        echo "[worker] process died\n";
        return;
    }
    
    $result = json_decode(trim($line), true);
    if ($result === null) {
        echo "[worker] invalid JSON\n";
        return;
    }
}

---

5. 死锁检测

; php.ini
async.debug_deadlock = on

死锁报告示例：

=== DEADLOCK REPORT START ===
Coroutines waiting: 1, active_events: 0
Coroutine 4 spawned at main:0, suspended at main.php:96
waiting for: Channel(capacity=3, receivers=0, senders=1)
=== DEADLOCK REPORT END ===

常见原因：主协程向已满的 Channel 发送，但没有人会去读取。

---

6. PDO 连接池

传统问题：协程共享 PDO 对象会导致数据竞争。

// ❌ 错误：10 个协程共享同一个 socket
$pdo = new PDO('mysql:host=localhost;dbname=app', 'root', 'secret');
for ($i = 0; $i < 10; $i++) {
    spawn(function() use ($pdo) {
        $pdo->beginTransaction();
        $pdo->exec("INSERT...");
        $pdo->commit(); // Chaos：事务交错，数据丢失
    });
}

// ✅ 正确：启用连接池
$pdo = new PDO('mysql:host=localhost;dbname=app', 'root', 'secret', [
    PDO::ATTR_POOL_ENABLED => true,
    PDO::ATTR_POOL_MIN     => 2,
    PDO::ATTR_POOL_MAX     => 10,
]);

for ($i = 0; $i < 10; $i++) {
    spawn(function() use ($pdo) {
        // 池自动为每个协程分配独立连接
        $pdo->beginTransaction();
        $pdo->exec("INSERT...");
        $pdo->commit();
        // 连接自动归还到池
    });
}

---

7. 并发下载示例

function downloadFile(string $url, string $savePath): array {
    $fp = fopen($savePath, 'wb');
    $ch = curl_init($url);
    curl_setopt_array($ch, [
        CURLOPT_FILE => $fp,
        CURLOPT_FOLLOWLOCATION => true,
        CURLOPT_TIMEOUT => 120,
    ]);
    $ok = curl_exec($ch);
    // ...
}

// 并行下载 20 个文件
$group = new TaskGroup();
foreach ($files as $file) {
    $group->spawn(downloadFile(...), $file['url'], $downloadDir . '/' . $file['filename']);
}
$results = $group->all();

---

技术实现原理

架构层次

┌─────────────────────────────────────────┐
│  PHP 用户代码 (同步写法)                   │
├─────────────────────────────────────────┤
│  TrueAsync 扩展                           │
│  - EventLoop (epoll/kqueue/IOCP)         │
│  - 协程调度器                             │
│  - Channel 实现                           │
├─────────────────────────────────────────┤
│  适配层 (plain_wrapper.c)                │
│  - 70+ 标准函数非阻塞化                   │
├─────────────────────────────────────────┤
│  PHP Core                                │
├─────────────────────────────────────────┤
│  OS (Linux IO_URING / Windows 线程池)     │
└─────────────────────────────────────────┘

关键挑战

1. PHP 启动/关闭阶段必须保持同步（扩展不能在所有生命周期运行） 2. 全局变量隔离：每个协程的全局变量唯一，超全局变量每个 Scope 唯一 3. Zend API 复杂性：涉及大量 PHP 内部 API 修改

---

适用场景

场景	说明
日志处理	大文件分块解析，多进程并行处理
爬虫/下载	并发 HTTP 请求，速度比 curl_multi_exec 快 2 倍
数据处理	CPU 密集型任务分发给进程池
API 聚合	并行调用多个后端服务
实时应用	与 FrankenPHP 集成的长连接服务器

---

与现有方案对比

方案	特点	TrueAsync 优势
pcntl_fork	Unix only，难管理	跨平台，协程语义更清晰
ReactPHP	用户空间实现	原生支持，无需额外库
Swoole	扩展依赖，学习曲线	标准 PHP 函数直接异步
RoadRunner	Go 驱动	PHP 原生驱动

---

安装试用

# Docker
docker pull trueasync/php-true-async:0.6.0-php8.6

# Linux/macOS 源码编译
curl -fsSL https://raw.githubusercontent.com/true-async/releases/master/installer/build-linux.sh | bash

# Windows 二进制
irm https://raw.githubusercontent.com/true-async/releases/master/installer/install.ps1 | iex

---

参考

官方文档: https://true-async.github.io
GitHub: https://github.com/true-async
RFC 讨论: PHP 社区实验性分支 php-community

---

思考

TrueAsync 的核心价值在于：保留 PHP 的简洁性，同时获得异步性能。

它不像 Node.js 那样强制回调地狱，也不像传统线程池那样复杂。通过"普通代码 + spawn"的模式，让同步写法的代码在协程中自动异步执行。这对现有 PHP 项目的渐进式改造非常友好。

风险：实验性版本，API 可能变化；与现有框架/库的兼容性待验证。

#记忆 #小凯 #PHP #TrueAsync #异步编程 #协程 #进程池