本文目录导读：
在PHP项目中实现任务中心（Task Center）通常有两种主要方式：同步任务队列（如数据库驱动的简易队列）和异步任务中心（如使用消息队列或任务调度系统），以下从架构设计、实现方案和代码示例三个层面详细说明：
架构设计核心要素
- 任务存储：MySQL/Redis/消息队列（如RabbitMQ、Beanstalkd）
- 任务状态机：待处理 → 处理中 → 失败/成功
- 执行机制：同步（实时执行）/ 异步（后台Worker）
- 任务重试：失败任务自动重试机制（指数退避）
- 监控报警：失败任务通知、执行耗时统计
轻量级方案：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);
}
最佳实践建议
- 任务幂等性：确保同一任务多次执行结果一致（通过唯一业务ID去重）
- 任务超时控制：PHP执行脚本设置
set_time_limit(0)，Worker内设置pcntl_alarm() - 优雅关闭：Worker捕获SIGTERM信号完成当前任务后退出
- 任务日志：记录每个任务的完整执行轨迹（开始时间、结束时间、结果、异常信息）
- 资源限制：每个Worker设置内存限制（
memory_limit），防止单个任务耗尽资源
推荐生产级组件
| 组件 | 类型 | 适用场景 |
|---|---|---|
| Laravel Queue | 框架内置 | 使用Laravel的项目 |
| Symfony Messenger | 框架内置 | 使用Symfony的项目 |
| Beanstalkd | 轻量级队列 | 中等规模，需要优先级 |
| RabbitMQ | 消息队列 | 高可靠、复杂路由 |
| Redis | 内存数据库 | 高速、简单任务 |
| SQS | 云服务 | AWS生态 |
选择方案时需综合考虑：并发量、数据可靠性要求、运维成本、团队技术栈，对于大多数PHP项目，MySQL + cron 方案足以应对初期需求,当遇到性能瓶颈时再平滑迁移到Redis或消息队列方案。
