README

基于 EasySwoole 自定义进程和 Unix Socket 实现的任务组件，支持同步与异步投递、按负载选择 Worker、并发限制、任务过期控制和运行状态统计。

环境要求

PHP 8.1+
Swoole 5.1+

安装

composer require easyswoole/task

定义任务

所有任务都必须实现 TaskInterface：

<?php

use EasySwoole\Task\AbstractInterface\TaskInterface;

final class ExampleTask implements TaskInterface
{
    public function __construct(private readonly string $name = 'EasySwoole')
    {
    }

    public function run(int $taskId, int $workerIndex): mixed
    {
        return sprintf(
            'hello %s, taskId=%d, workerIndex=%d',
            $this->name,
            $taskId,
            $workerIndex
        );
    }

    public function onException(
        Throwable $throwable,
        int $taskId,
        int $workerIndex
    ): mixed {
        // 记录日志或执行任务级异常处理
        return null;
    }
}

任务实例会随 Package 使用 PHP 原生序列化传递给 Worker，因此任务属性也会一并保留。任务对象及其属性必须可序列化，不要保存闭包、连接、文件句柄等无法序列化的运行时资源。

必须传入任务实例，不支持传递类名。任务需要构造参数时，应先在应用层完成实例化：

$task->async(new ExampleTask('EasySwoole'));

基本使用

<?php

use EasySwoole\Task\Task;

$task = new Task();
$task->attachToServer($server);

// 异步投递，成功返回任务 ID；失败时返回 TaskStatus 枚举
$taskId = $task->async(new ExampleTask('async'));

// 同步投递，等待并返回任务执行结果
$result = $task->sync(new ExampleTask('sync'));

Task 实例只能附加到一个 Swoole Server 一次，并且必须在 Server 启动前调用 attachToServer()。

投递超时

pushTimeOut 表示整个任务投递流程的最大等待时间。序列化、连接 Worker、发送任务包和等待投递响应会共享同一个截止时间：

$taskId = $task->async(new ExampleTask(), 1.5);
$result = $task->sync(new ExampleTask(), 3.0);

异步方法未传入超时时间时使用配置中的默认值；同步方法的默认值为 3 秒。超时时间必须是有限数值，传入 NAN 或 INF 会抛出 InvalidArgumentException；小于等于 0 表示不设置投递截止时间和任务过期时间。

幂等提醒： Worker 会在接收任务及真正开始执行前检查过期时间，尚未执行的超时任务会被直接丢弃。但投递超时并不等同于取消任务；任务一旦开始运行，PHP 无法安全地强制中断它，即使调用方已经超时，任务仍可能继续完成。调用方可能重试的写入、扣款、通知等业务必须自行保证幂等性。

Worker 选择与繁忙处理

客户端会读取共享状态表，从尚未达到 maxRunningNum 的 Worker 中选择 running 最小的一个投递；多个 Worker 负载相同时随机选择。

一次调用只会向一个 Worker 投递一次：

不使用任务队列。
不会因连接失败、超时或 Worker 竞态繁忙而自动改投其他 Worker。
没有可用 Worker，或者选中的 Worker 在投递期间达到并发上限时，返回 TaskStatus::ERROR_PROCESS_BUSY。
连接、发送或接收失败时返回 TaskStatus::ERROR_SOCK_TIMEOUT，调用方自行决定是否重试。

异步投递返回任务 ID 仅表示 Worker 已经接收该任务，不表示任务已经执行完成。

Worker 退出与任务编码注意事项

Worker 收到 SIGTERM 后采用优雅退出：停止监听并拒绝新任务，已经开始的任务继续执行；最后一个任务结束后清理共享状态记录和 Unix Socket 文件，然后进程正常退出。组件不会强制取消已经开始的业务任务。

因此，任务实现必须能够在有限时间内自行结束：

不要编写没有退出条件的 while (true) 等无限循环。
不要执行永久阻塞的网络、文件或进程调用。
CPU 密集循环不会主动让出协程调度，可能导致信号和其他协程无法及时处理。
长循环应设置明确的结束条件，并在适当位置使用协程 I/O 或 Coroutine::sleep() 让出执行权。
仅添加 Coroutine::sleep() 不能让一个永不结束的任务支持优雅退出；任务本身仍然必须具备退出条件。

例如：

use Swoole\Coroutine;

while (!$this->shouldStop()) {
    $this->processBatch();
    Coroutine::sleep(0.001);
}

如果任务发生 CPU 死循环或永久阻塞，Worker 将无法完成优雅退出。这属于任务代码问题，组件不会通过强制取消协程来掩盖它。

配置

配置必须在调用 attachToServer() 前完成：

$config = $task->getConfig();

$config->setWorkerNum(4);
$config->setMaxRunningNum(128);
$config->setMaxPackageSize(2 * 1024 * 1024);
$config->setPushTimeOut(5.0);
$config->setServerName('MyApplication');
$config->setTempDir('/tmp');

// 可选任务白名单；每次注册一个 TaskInterface 实现类
$config->addRegistryTask(ExampleTask::class);

$config->setOnWorkerStart(
    static function (\EasySwoole\Task\Worker $worker): void {
        // 每个 Worker 启动后执行一次，可在这里初始化 Worker 级资源
    }
);

$config->setOnException(
    static function (Throwable $throwable, int $workerIndex): void {
        // Worker 级异常处理
    }
);

配置	默认值	说明
`workerNum`	`3`	Task Worker 数量，范围为 1～512
`maxRunningNum`	`128`	每个 Worker 的最大并发任务数
`maxPackageSize`	`2 MiB`	Socket 完整帧最大大小（包含 4 字节长度头，最小 512 字节）
`pushTimeOut`	`5.0`	异步投递的默认等待时间
`serverName`	`EasySwoole`	Worker 进程名和 Socket 文件名前缀
`tempDir`	当前工作目录	Unix Socket 文件目录
`taskRegistry`	`[]`	可选任务类白名单；空数组表示不限制
`onWorkerStart`	`null`	Worker 启动回调；每个 Worker 执行一次并接收当前 `Worker` 实例
`onException`	`null`	Worker 级异常处理回调

onWorkerStart 在 Worker 状态初始化后、开始监听任务前调用，适合初始化数据库连接、客户端等进程内资源。回调在每个 Worker 进程中分别执行；如果回调抛出异常，该 Worker 将启动失败。

完整 Unix Socket 路径由 tempDir、serverName 和 Worker ID 组成，按字节计算不能超过 100 字节；路径过长会在创建 Worker 前抛出 InvalidArgumentException。项目目录较深时建议将 tempDir 设置为 /tmp 等短路径。

状态码

组件状态码定义在 TaskStatus 枚举中：

use EasySwoole\Task\TaskStatus;

$status = TaskStatus::ERROR_PROCESS_BUSY;

echo $status->value;     // -2
echo $status->message(); // task process busy

if ($result instanceof TaskStatus) {
    echo $result->message();
}

枚举	值	说明
`ERROR_PROCESS_BUSY`	`-2`	Worker 繁忙
`ERROR_PROTOCOL_ERROR`	`-3`	协议错误
`ERROR_ILLEGAL_PACKAGE`	`-4`	非法任务包
`ERROR_TASK_ERROR`	`-5`	任务执行异常
`ERROR_PACKAGE_EXPIRE`	`-6`	任务包已过期
`ERROR_SOCK_TIMEOUT`	`-7`	Socket 等待超时
`ERROR_PACKAGE_TOO_LARGE`	`-8`	任务包超过大小限制
`ERROR_RESPONSE_TOO_LARGE`	`-9`	同步任务返回值超过大小限制

协议响应是包含版本号、成功标记、状态码和业务数据的序列化数组。公开 API 对失败状态返回 TaskStatus 枚举，业务返回的负整数仍是普通 int，两者不会混淆。组件不负责阻止重复投递；需要幂等的任务应由应用层按业务主键处理。

同步任务返回值也通过 PHP 原生序列化传输。无法序列化的返回值会被视为任务错误；编码后的响应超过 maxPackageSize 时返回 ERROR_RESPONSE_TOO_LARGE。

查看 Worker 状态

$status = $task->status();

每个 Worker 的状态包含：

accepting：是否接受新任务；退出中的 Worker 为 0，不会再被客户端选中
running：正在执行的任务数
success：成功任务数
fail：失败或被丢弃的任务数
pid：Worker 进程 ID
startUpTime：Worker 启动时间戳

运行测试

composer test

设计实现

进程与共享状态

Task 在 Swoole Server 启动前创建一个共享 Swoole\Table 和一个共享任务 ID 原子计数器，然后按照 workerNum 创建独立的 Task Worker 进程。

每个 Worker 拥有独立的 Unix Domain Socket，文件名格式为：

{tempDir}/{serverName}.TaskWorker.{workerId}.sock

完整路径按字节计算最多允许 100 字节。macOS 的 sockaddr_un.sun_path 通常为 104 字节，Linux 通常为 108 字节；限制为 100 字节是为了给结尾空字符和不同系统实现预留空间。

共享状态表中的每一行对应一个 Worker：

accepting   是否接受新任务
running     当前执行中的任务数
success     成功任务数
fail        失败任务数
pid         Worker 进程 ID
startUpTime Worker 启动时间

Worker 选择

客户端投递任务前读取共享状态表：

排除尚未启动、已经退出或 accepting != 1 的 Worker。
排除 running >= maxRunningNum 的 Worker。
在剩余 Worker 中选择 running 最小的一个。
多个 Worker 负载相同时随机选择。

状态表是负载快照，多个客户端仍可能同时选中同一个 Worker。因此 Worker 收包后还会使用 Swoole\Table::incr() 原子增加 running，再次检查并发上限。竞态下超过上限的请求直接返回 ERROR_PROCESS_BUSY。

组件不会排队，也不会自动改投其他 Worker。这样可以避免调用方不知道任务是否已经送达时发生隐式二次执行。

投递协议

任务实例被封装为 Package，使用 PHP 原生序列化传输。每一帧由 4 字节网络字节序长度头和序列化数据组成：

+----------------------+-----------------------+
| 4 字节数据长度       | 序列化 Package/响应   |
+----------------------+-----------------------+

客户端和 Worker 都使用 maxPackageSize 限制完整帧大小。Worker 会先验证长度头，再接收包体、反序列化、检查包类型、验证 TaskInterface，最后执行可选任务注册表校验。

响应使用结构化数组区分状态码和业务结果，因此业务返回负整数时不会被误认为 TaskStatus。

同步与异步任务

同步任务在 Worker 中执行完成后，将业务结果序列化并通过原连接返回。返回帧超过 maxPackageSize 时返回 ERROR_RESPONSE_TOO_LARGE。

异步任务在 Worker 获取并发槽位后先返回任务 ID，然后继续在当前 Worker 协程中执行。任务 ID 只表示 Worker 已经接收任务，不表示任务执行成功。

任务执行抛出异常时，首先调用任务自己的 TaskInterface::onException()：

任务级异常处理正常返回时，其返回值可以作为同步结果，但任务计入 fail。
任务级异常处理继续抛出时，异常到达最外层并触发一次 Worker 级 onException 回调。

投递截止时间

pushTimeOut 会在调用开始时转换为绝对截止时间并写入任务包。序列化、连接、发送和等待响应共享同一个时间预算，每一步都使用剩余时间。

Worker 在收到任务后和真正开始执行前都会检查截止时间。尚未执行但已经过期的任务会被拒绝；已经开始执行的任务不会因为客户端超时而被强制取消。

优雅退出

Worker 收到 SIGTERM 后按照以下顺序退出：

将共享状态中的 accepting 设置为 0，客户端立即停止向它投递新任务。
关闭监听 Unix Socket。
删除 Unix Socket 文件。
没有运行中任务时立即删除状态行。
有运行中任务时等待任务自然完成，由最后一个任务释放并发槽位时删除状态行。

easyswoole / task

Package info

Security

Statistics