怎样在PHP项目中实现任务中心？

本文目录导读：

1. 架构设计核心要素
2. 轻量级方案：MySQL驱动 + cron
3. 企业级方案：Redis + 消息队列
4. 高级特性实现
5. 监控与运维
6. 最佳实践建议
7. 推荐生产级组件

在PHP项目中实现任务中心（Task Center）通常有两种主要方式：同步任务队列（如数据库驱动的简易队列）和异步任务中心（如使用消息队列或任务调度系统），以下从架构设计、实现方案和代码示例三个层面详细说明：

架构设计核心要素

1. 任务存储：MySQL/Redis/消息队列（如RabbitMQ、Beanstalkd）
2. 任务状态机：待处理 → 处理中 → 失败/成功
3. 执行机制：同步（实时执行）/ 异步（后台Worker）
4. 任务重试：失败任务自动重试机制（指数退避）
5. 监控报警：失败任务通知、执行耗时统计

轻量级方案：MySQL驱动 + cron

适合中小型项目,无需额外依赖。

数据表设计

CREATE TABLE `tasks` (
    `id` INT AUTO_INCREMENT PRIMARY KEY,
    `type` VARCHAR(50) NOT NULL COMMENT '任务类型',
    `payload` JSON NOT NULL COMMENT '任务参数',
    `status` ENUM('pending','processing','success','failed') DEFAULT 'pending',
    `retry_count` TINYINT DEFAULT 0 COMMENT '已重试次数',
    `max_retries` TINYINT DEFAULT 3,
    `priority` TINYINT DEFAULT 0 COMMENT '优先级 0-10',
    `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    `updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    `scheduled_at` TIMESTAMP NULL COMMENT '计划执行时间',
    `error_info` TEXT NULL,
    INDEX `status_scheduled` (`status`, `scheduled_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

核心任务类

class TaskCenter {
    private PDO $db;
    private array $handlers = [];
    public function __construct(PDO $db) {
        $this->db = $db;
    }
    // 注册任务处理器
    public function registerHandler(string $type, callable $handler): void {
        $this->handlers[$type] = $handler;
    }
    // 创建任务
    public function createTask(string $type, array $payload, int $priority = 0, 
                              ?DateTime $scheduleAt = null): int {
        $stmt = $this->db->prepare(
            "INSERT INTO tasks (type, payload, priority, scheduled_at) 
             VALUES (?, ?, ?, ?)"
        );
        $stmt->execute([
            $type,
            json_encode($payload),
            $priority,
            $scheduleAt?->format('Y-m-d H:i:s')
        ]);
        return (int)$this->db->lastInsertId();
    }
    // 处理单个任务（同步执行）
    public function processTask(int $taskId): bool {
        $this->db->beginTransaction();
        try {
            $task = $this->db->query("SELECT * FROM tasks WHERE id = $taskId FOR UPDATE")->fetch();
            if (!$task || $task['status'] !== 'pending') return false;
            // 锁定任务
            $this->updateStatus($taskId, 'processing');
            $this->db->commit();
            // 执行处理
            $handler = $this->handlers[$task['type']] ?? null;
            if (!$handler) throw new Exception("No handler for type: {$task['type']}");
            $result = $handler(json_decode($task['payload'], true));
            // 处理成功
            $this->updateStatus($taskId, 'success');
            return true;
        } catch (Exception $e) {
            $this->db->rollBack();
            $this->handleTaskFailure($taskId, $e);
            return false;
        }
    }
    // 批量处理待办任务（供cron调用）
    public function processPendingTasks(int $batchSize = 10): array {
        $results = [];
        $tasks = $this->db->query(
            "SELECT id FROM tasks 
             WHERE status = 'pending' 
               AND (scheduled_at IS NULL OR scheduled_at <= NOW())
             ORDER BY priority DESC, created_at ASC 
             LIMIT $batchSize FOR UPDATE SKIP LOCKED"
        )->fetchAll();
        foreach ($tasks as $task) {
            $results[$task['id']] = $this->processTask((int)$task['id']);
        }
        return $results;
    }
    private function updateStatus(int $taskId, string $status, ?string $error = null): void {
        $this->db->prepare(
            "UPDATE tasks SET status = ?, error_info = ?, updated_at = NOW() WHERE id = ?"
        )->execute([$status, $error, $taskId]);
    }
    private function handleTaskFailure(int $taskId, Exception $e): void {
        $task = $this->db->query("SELECT * FROM tasks WHERE id = $taskId")->fetch();
        if ($task['retry_count'] < $task['max_retries']) {
            // 指数退避重试
            $delay = pow(2, $task['retry_count']) * 60; // 1分钟, 2分钟, 4分钟...
            $this->db->prepare(
                "UPDATE tasks SET status = 'pending', retry_count = retry_count + 1, 
                 error_info = ?, scheduled_at = DATE_ADD(NOW(), INTERVAL ? SECOND) 
                 WHERE id = ?"
            )->execute([$e->getMessage(), $delay, $taskId]);
        } else {
            $this->updateStatus($taskId, 'failed', $e->getMessage());
            // 触发报警
            $this->notifyFailure($taskId);
        }
    }
    private function notifyFailure(int $taskId): void {
        // 实现邮件、钉钉、企业微信通知
        // mail(...) 或 调用第三方API
    }
}

使用示例

// 初始化任务中心
$db = new PDO('mysql:host=localhost;dbname=test', 'root', '');
$taskCenter = new TaskCenter($db);
// 注册处理程序
$taskCenter->registerHandler('send_email', function($payload) {
    return mail($payload['to'], $payload['subject'], $payload['body']);
});
$taskCenter->registerHandler('generate_report', function($payload) {
    // 生成报表逻辑
    return generateDailyReport($payload['date']);
});
// 创建任务
$taskId = $taskCenter->createTask('send_email', [
    'to' => 'user@example.com',
    'subject' => 'Welcome',
    'body' => 'Thank you for registering'
], priority: 5);
// 创建定时任务
$taskCenter->createTask('generate_report', 
    ['date' => '2024-01-15'],
    scheduleAt: new DateTime('2024-01-15 02:00:00')
);

Cron脚本（每30秒执行一次）

// cron_process.php
require 'vendor/autoload.php';
$taskCenter = new TaskCenter(getDbConnection());
try {
    $result = $taskCenter->processPendingTasks(10);
    echo "Processed " . count($result) . " tasks\n";
} catch (Exception $e) {
    error_log("Task processing error: " . $e->getMessage());
}

企业级方案：Redis + 消息队列

适合高并发、需要可靠异步处理的场景。

使用Redis作任务队列

class AsyncTaskCenter {
    private Redis $redis;
    private string $queuePrefix = 'task:';
    public function __construct(Redis $redis) {
        $this->redis = $redis;
    }
    // 推送任务到队列
    public function dispatch(string $jobClass, array $data, 
                           string $queue = 'default', ?int $delay = null): void {
        $payload = json_encode([
            'class' => $jobClass,
            'data' => $data,
            'attempts' => 0,
            'created_at' => time()
        ]);
        if ($delay) {
            // 延迟队列使用ZSet
            $this->redis->zAdd("{$this->queuePrefix}delayed:$queue", 
                              time() + $delay, $payload);
        } else {
            // 即时队列使用List
            $this->redis->lPush("{$this->queuePrefix}$queue", $payload);
        }
    }
    // Worker进程获取任务
    public function getJob(string $queue = 'default', int $timeout = 5): ?array {
        // 先检查延期队列
        $delayedKey = "{$this->queuePrefix}delayed:$queue";
        $ready = $this->redis->zRangeByScore($delayedKey, 0, time(), ['limit' => [0, 1]]);
        if ($ready) {
            $this->redis->zRem($delayedKey, $ready[0]);
            return ['payload' => json_decode($ready[0], true), 'queue' => $queue];
        }
        // 从主队列获取
        $job = $this->redis->brPop(["{$this->queuePrefix}$queue"], $timeout);
        return $job ? ['payload' => json_decode($job[1], true), 'queue' => $job[0]] : null;
    }
    // 失败任务处理（重试/死信）
    public function failJob(array $job, string $error, int $maxAttempts = 3): void {
        $payload = $job['payload'];
        $payload['attempts']++;
        if ($payload['attempts'] < $maxAttempts) {
            // 指数退避重试
            $delay = pow(2, $payload['attempts']) * 10; // 10秒开始
            $this->redis->zAdd("{$this->queuePrefix}delayed:{$job['queue']}",
                              time() + $delay, json_encode($payload));
        } else {
            // 进入死信队列
            $this->redis->lPush("{$this->queuePrefix}dead_letter", json_encode([
                'payload' => $payload,
                'error' => $error,
                'failed_at' => time()
            ]));
            // 触发告警
        }
    }
}

Worker守护进程（使用Supervisor管理）

// worker.php
require 'vendor/autoload.php';
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$taskCenter = new AsyncTaskCenter($redis);
while (true) {
    $job = $taskCenter->getJob('default', 60);
    if (!$job) continue;
    try {
        $class = $job['payload']['class'];
        $handler = new $class();
        $handler->handle($job['payload']['data']);
        echo "[" . date('Y-m-d H:i:s') . "] Processed job: $class\n";
    } catch (Exception $e) {
        $taskCenter->failJob($job, $e->getMessage());
        echo "[" . date('Y-m-d H:i:s') . "] Job failed: {$e->getMessage()}\n";
    }
}

Supervisor配置示例

[program:task-worker]
command=php /path/to/worker.php
user=www-data
autostart=true
autorestart=true
startretries=3
numprocs=4
process_name=%(program_name)s_%(process_num)02d
stderr_logfile=/var/log/task_worker.err.log
stdout_logfile=/var/log/task_worker.out.log

高级特性实现

任务链（Chain Tasks）

class TaskChain {
    public function createChain(array $services, $initialData): void {
        $prevId = null;
        foreach ($services as $index => $service) {
            $data = [
                'service' => $service,
                'input' => $index === 0 ? $initialData : null,
                'prev_task_id' => $prevId
            ];
            $prevId = $this->dispatch('chain_handler', $data);
        }
    }
}

任务进度跟踪

// 任务处理时更新进度
public function updateProgress(int $taskId, int $progress, string $message = ''): void {
    $this->redis->hSet("task:progress:$taskId", 'progress', $progress);
    $this->redis->hSet("task:progress:$taskId", 'message', $message);
    $this->redis->expire("task:progress:$taskId", 3600);
}
// 任务查询
public function getProgress(int $taskId): array {
    return $this->redis->hGetAll("task:progress:$taskId") ?: ['progress' => 0];
}

任务优先级队列（Redis实现）

// 使用多个队列按优先级分离
public function dispatchByPriority(string $jobClass, array $data, int $priority): void {
    $queue = $priority >= 5 ? 'high' : ($priority >= 3 ? 'medium' : 'low');
    $this->dispatch($jobClass, $data, $queue);
}
// Worker按权重消费
public function getJobWithPriority(array $queues = ['high', 'medium', 'low']): ?array {
    foreach ($queues as $queue) {
        $job = $this->redis->lPop("{$this->queuePrefix}$queue");
        if ($job) {
            return ['payload' => json_decode($job, true), 'queue' => $queue];
        }
    }
    return null;
}

监控与运维

任务统计看板

class TaskMonitor {
    public function getStats(): array {
        return [
            'total_today' => $this->db->query(
                "SELECT COUNT(*) FROM tasks WHERE DATE(created_at) = CURDATE()"
            )->fetchColumn(),
            'pending' => $this->db->query(
                "SELECT COUNT(*) FROM tasks WHERE status = 'pending'"
            )->fetchColumn(),
            'failed_last_hour' => $this->db->query(
                "SELECT COUNT(*) FROM tasks WHERE status = 'failed' 
                 AND updated_at >= DATE_SUB(NOW(), INTERVAL 1 HOUR)"
            )->fetchColumn(),
            'avg_execution_time' => $this->db->query(
                "SELECT AVG(TIMESTAMPDIFF(SECOND, created_at, updated_at)) 
                 FROM tasks WHERE status = 'success' AND updated_at >= DATE_SUB(NOW(), INTERVAL 1 HOUR)"
            )->fetchColumn()
        ];
    }
}

失败任务报警

// 使用Webhook推送通知
public function alertFailedTask(array $task): void {
    $message = sprintf(
        "Task #%s failed after %d attempts: %s",
        $task['id'],
        $task['retry_count'],
        $task['error_info']
    );
    // 钉钉Webhook
    $this->sendDingTalk($message);
    // 或发送邮件给运维团队
    mail('ops@company.com', 'Task Failed', $message);
}

最佳实践建议

1. 任务幂等性：确保同一任务多次执行结果一致（通过唯一业务ID去重）
2. 任务超时控制：PHP执行脚本设置set_time_limit(0)，Worker内设置pcntl_alarm()
3. 优雅关闭：Worker捕获SIGTERM信号完成当前任务后退出
4. 任务日志：记录每个任务的完整执行轨迹（开始时间、结束时间、结果、异常信息）
5. 资源限制：每个Worker设置内存限制（memory_limit），防止单个任务耗尽资源

推荐生产级组件

选择方案时需综合考虑：并发量、数据可靠性要求、运维成本、团队技术栈，对于大多数PHP项目，MySQL + cron 方案足以应对初期需求,当遇到性能瓶颈时再平滑迁移到Redis或消息队列方案。