README

Crontab 是 FiberPHP 框架的定时任务子包，基于 Workerman 多进程架构，支持秒级精度调度、协程执行、实时日志监控和 Redis 分布式锁。

核心特性

秒级精度调度：内置 6/7 位 Crontab 表达式解析器（含秒字段），精确到秒
多进程隔离：调度器、执行器、WebSocket 日志服务三个独立进程，互不干扰
协程支持：enable_coroutine=true 时使用 Workerman Coroutine 执行 I/O 密集型任务
实时日志推送：WebSocket 服务订阅 Redis channel，浏览器端实时可见执行日志
分布式锁：Redis SET NX EX 分布式锁（随机 token + Lua 原子释放），多实例部署任务不重复执行
多任务类型：Console Command / Class 静态方法 / URL HTTP 请求（支持 method/headers/body/timeout）/ Shell 子进程
失败重试：max_retry 控制退避重试次数（线性退避），重试耗尽自动 POST webhook 告警
成功/失败 Ping：ping_success_url / ping_failure_url（GET healthcheck 类服务），失败 ping 仅最终失败时发出
日志保留：log_retention_days 自动清理过期执行日志
平滑重启：onStop 标志位 + Workerman 信号机制，确保任务不丢失

环境要求

PHP >= 8.3
ext-redis（Redis 驱动必备）
FiberPHP Framework
Workerman >= 5.1
fiberphp/database（内置 mysql/pgsql/sqlite 驱动）用于存放任务表

安装

composer require fiberphp/crontab

PackageInstaller::discover 会自动把以下文件拷贝到应用（存在时不覆盖）：

config/crontab.php          ← 顶层配置（enable / scan_interval / 表名 / 队列 / 锁 / 日志）
config/process/crontab.php  ← 三个 crontab Worker 进程声明

导入数据库表：

mysql -u root -p your_database < vendor/fiberphp/crontab/crontab.sql

配置

config/crontab.php：

return [
    'enable'           => true,
    'enable_coroutine' => true,                   // Workerman Coroutine 执行器 or Timer 异步
    'scan_interval'    => 5,                      // 调度器扫描间隔（秒）
    'table_cron'       => 'crontab',              // 任务表
    'table_log'        => 'crontab_log',          // 执行日志表
    'cron_queue'       => 'crontab:cron_queue',   // Redis 就绪队列（LPUSH / BRPOP）
    'lock_prefix'      => 'crontab:lock:',        // 分布式锁 KEY 前缀
    'log_subscribe'    => 'crontab:logs',         // 实时日志 Redis Pub/Sub 频道
    'log_write_file'   => false,                  // 同时写 PSR-3 logger
    'retry_prefix'     => 'crontab:retry:',       // 失败重试计数 KEY 前缀
    'retry_backoff'    => 60,                     // 重试退避基数（秒），第 n 次失败后延迟 n * retry_backoff
    'alert_webhook'    => '',                     // 全局失败告警 webhook（任务级 webhook 字段优先）
    'log_retention_days' => 30,                   // 执行日志保留天数，0 永久保留
];

config/process/crontab.php（Install 注入后可自行修改端口/进程数）：

return [
    'scheduler' => [
        'handler'     => \FiberPHP\Crontab\Process\CronScheduler::class,
        'listen'      => 'text://0.0.0.0:12346',   // TCP 控制端口，供 Client 调用
        'count'       => 1,
    ],
    'exec' => [
        'handler'     => \FiberPHP\Crontab\Process\CronExec::class,
        'listen'      => 'text://0.0.0.0:12347',
        'count'       => 1,
    ],
    'websocket' => [
        'handler'     => \FiberPHP\Crontab\Process\LogSocket::class,
        'listen'      => 'websocket://0.0.0.0:12348',
        'count'       => 1,
    ],
];

添加任务（无需重启服务）

直接写入 crontab 数据表即可，CronScheduler 每 scan_interval 秒扫描一次：

INSERT INTO `crontab`
(`name`, `cron_expression`, `task_type`, `command`, `status`)
VALUES
('清理缓存',     '0 0 2 * * *',  2,  'App\\Task\\ClearCache::run',                 0),
('调用报表 URL', '*/30 * * * *', 3,  'https://app.test/report/generate',           0),
('每日对账',     '0 5 0 * * *',  1,  'fiber reconcile:daily',                      0),
('清理临时目录', '0 10 3 * * 0', 4,  'rm -rf /tmp/app-*',                           0);

task_type：1 Console Command · 2 Class::method · 3 URL · 4 Shell。

URL 任务 command 兼容两种写法——纯 URL（GET）或 JSON 增强格式：

{"url":"https://api.test/notify","method":"POST","headers":{"X-Token":"abc"},"body":{"k":"v"},"timeout":10}

失败重试与告警：max_retry（额外重试次数，0 不重试）失败后按 n * retry_backoff 秒退避重投；重试耗尽或未启用重试时，向任务级 webhook（空则回退全局 alert_webhook）POST 告警 JSON（含 task_id/task_name/log_id/attempt/output/failed_at）。

成功/失败 Ping（healthcheck 类服务，GET 请求）：每次成功执行后 GET ping_success_url；仅最终失败时 GET ping_failure_url（中间重试不 ping，与告警语义一致）。

已部署环境升级（新增列）：

ALTER TABLE `crontab` ADD COLUMN `max_retry` int(11) DEFAULT 0 AFTER `lock_time`;
ALTER TABLE `crontab` ADD COLUMN `webhook` varchar(500) DEFAULT NULL AFTER `max_retry`;
ALTER TABLE `crontab` ADD COLUMN `ping_success_url` varchar(500) DEFAULT NULL AFTER `webhook`;
ALTER TABLE `crontab` ADD COLUMN `ping_failure_url` varchar(500) DEFAULT NULL AFTER `ping_success_url`;
ALTER TABLE `crontab_log` ADD COLUMN `attempt` int(11) DEFAULT 1 AFTER `pid`;

代码操作（通过 TCP 客户端与调度器进程通信）：

use FiberPHP\Crontab\Client;

$client = new Client();
// 创建 / 列表 / 日志 / 立即执行 / 启动 / 暂停 / 重启 / 编辑
$client->request(['method' => 'createTask',          'args' => ['name' => 'test', /* ... */]]);
$client->request(['method' => 'getList',             'args' => ['paginate' => ['page' => 1, 'list_rows' => 20]]]);
$client->request(['method' => 'getTaskLogs',         'args' => ['where' => ['crontab_id' => 1]]]);
$client->request(['method' => 'executeImmediately',  'args' => ['id' => 1]]);
$client->request(['method' => 'reloadTask',          'args' => ['id' => 1]]);
$client->request(['method' => 'closeTask',           'args' => ['id' => 1]]);
$client->request(['method' => 'restartTask',         'args' => ['id' => 1]]);
$client->request(['method' => 'updateTask',          'args' => ['id' => 1, 'cron_expression' => '0 */5 * * * *']]);

助手函数（从容器拿 Client 单例）：

crontab()->request(['method' => 'getList', 'args' => []]);

自定义任务（Class 类型）

namespace App\Task;

class ClearCache
{
    public static function run(array $params = []): string
    {
        // 业务逻辑
        return '清理完成';
    }
}

数据库 command 填 App\Task\ClearCache::run，如需传参写 App\Task\ClearCache::run({"max": 100})（JSON）或传统位置参数形式。

实时日志监控

ws://0.0.0.0:12348 连接后：

{"type":"subscribe","channel":"crontab:logs"}

随后每条执行日志会以 JSON 推送：

{
  "type": "log",
  "channel": "crontab:logs",
  "data": {
    "task_id": 1,
    "log_id":  1023,
    "level":   "success",
    "message": "类方法执行成功: ...",
    "pid":     31424,
    "timestamp": 1766410111
  },
  "timestamp": 1766410111
}

架构

MySQL crontab 表
   │
   ▼ scan_interval 秒级扫描
CronScheduler 进程（text:12346 控制端口 + onMessage CRUD）
   │ LPUSH
   ▼
Redis List（crontab:cron_queue）
   │ BRPOP timeout=1
   ▼
CronExec 进程（CoroutineExec / AsyncTaskExec 四种任务类型）
   │ ├─ command/shell → symfony/process 子进程
   │ ├─ class         → ClassExec::execute 静态调用（支持 JSON/位置/常量表参数）
   │ └─ url           → Workerman\Http\Client
   │
   └─── PUBLISH ──► Redis Pub/Sub (crontab:logs)
                         │
                         ▼
                   LogSocket WebSocket（ws:12348）
                         │
                         ▼  浏览器 / API 客户端订阅

分布式调度原理

Crontab 天然支持多服务器部署（计算进程无状态 + 共享存储有状态），核心安全机制有三层：

第一层：CAS 防重复投递（Scheduler 层）

多 Scheduler 并发扫描同一张表时，CAS 原子更新保证只有一个实例成功推进 next_run_time：

┌── Server A Scheduler ──┐     ┌── Server B Scheduler ──┐
│ scan: task#1 next=10:00 │     │ scan: task#1 next=10:00 │
│                          │     │                          │
│ UPDATE crontab SET       │     │ UPDATE crontab SET       │
│   next_run_time=10:05    │     │   next_run_time=10:05    │
│   WHERE id=1             │     │   WHERE id=1             │
│   AND next_run_time=10:00│     │   AND next_run_time=10:00│
│   → 影响 1 行 ✓          │     │   → 影响 0 行 ✗          │
│                          │     │                          │
│ LPUSH queue ✓            │     │ 跳过投递 ✗               │
└──────────────────────────┘     └──────────────────────────┘

代码位于 TaskManager::advanceNextRunTime()：

$affected = db()->table($this->crontabTable())
    ->where('id', $taskId)
    ->where('next_run_time', $oldNextRunTime)   // CAS 关键条件
    ->update(['next_run_time' => $newNextRunTime]);

return $affected > 0;  // 只有推进成功的 Scheduler 才 LPUSH

第二层：Redis NX 锁防重复执行（Exec 层）

Scheduler 投递后，多个 Exec 进程（本机 + 其他机器）都能 BRPOP 到同一条任务。Redis 分布式锁保证只有一个 Exec 真正执行：

┌── Exec A ──┐    ┌── Exec B ──┐    ┌── Exec C ──┐
│ BRPOP ✓    │    │ BRPOP (等)  │    │ BRPOP (等)  │
│            │    │             │    │             │
│ SET lock   │    │             │    │             │
│ NX EX 300  │    │             │    │             │
│   → OK ✓   │    │             │    │             │
│            │    │             │    │             │
│ 执行任务   │    │ (不执行)    │    │ (不执行)    │
│            │    │             │    │             │
│ Lua 原子   │    │             │    │             │
│ 释放锁     │    │             │    │             │
└────────────┘    └─────────────┘    └─────────────┘

加锁：SET crontab:lock:{taskId} {randomToken} NX EX {lockTime}（NX = 不存在才设置）
释放：Lua 脚本比对 token 后 DEL，防止误删他人的锁
超时：lock_time 到期自动释放，防止 Exec 进程崩溃后锁永远不释放

第三层：Redis List 天然负载均衡（队列层）

Scheduler LPUSH 到同一个 Redis List，所有机器的 CronExec 进程 BRPOP 同一条队列——Redis 本身就是分布式队列，自然实现多消费者负载均衡：

         Scheduler A ──┐
                        ├──▶ Redis List (crontab:cron_queue)
         Scheduler B ──┘              │
                     ┌────────────────┼────────────────┐
                     ▼                ▼                ▼
                 Exec A            Exec B            Exec C
              (Server A)         (Server B)         (Server C)

共享存储前提

三层安全机制都依赖所有进程连接同一套 MySQL + Redis：

组件	用途	必须共享
MySQL	`crontab` 表（任务配置 + CAS 更新）、`crontab_log` 表（执行日志）	✅ 所有机器必须连同一个 DB
Redis	就绪队列（LPUSH/BRPOP）、分布式锁（SET NX EX）、重试计数器、实时日志 Pub/Sub	✅ 所有机器必须连同一个 Redis

已支持 vs 待增强

能力	当前状态	说明
多 Scheduler 安全	✅ 原生支持	CAS 防重复投递
多 Exec 负载均衡	✅ 原生支持	BRPOP + Redis NX 锁
多 Admin 安全	✅ 原生支持	Admin 无状态 HTTP，改共享 DB
Scheduler 高可用	⚠️ 依赖部署	加 ≥2 实例 + Nginx stream 负载均衡
Redis 高可用	⚠️ 依赖部署	Redis Sentinel / Cluster
MySQL 高可用	⚠️ 依赖部署	MySQL Master-Slave / ProxySQL
Admin 多 Scheduler 直连	⚠️ 需加 Nginx	见 crontab-panel 多服务器部署

License

MIT

fiberphp / crontab

Maintainers

Package info

Statistics

Security