PHP 怎么实现分布式定时

wen PHP项目 2026-08-05 245

本文目录导读：

PHP 怎么实现分布式定时

核心问题
主流实现方案
实现分布式定时任务的通用模式
生产环境最佳实践
总结推荐

在 PHP 中实现分布式定时任务，通常有以下几种方案，这里为你详细梳理，从简单到复杂：

核心问题

分布式定时任务要解决的三个核心问题：

任务分发：确保任务只被一台机器执行（避免重复执行）
任务调度：何时执行、执行频率
失败重试：任务执行失败后的处理

主流实现方案

方案1：基于 Redis 实现（推荐入门）

原理：利用 Redis 的 SETNX 或 ZSet 实现分布式锁，配合 brpoplpush 或 BLPOP 实现延迟队列。

// 1. 使用 SETNX 实现分布式锁
class DistributedLock {
    private $redis;
    private $lockKey;
    private $expire;
    public function __construct($redis, $lockKey, $expire = 10) {
        $this->redis = $redis;
        $this->lockKey = $lockKey;
        $this->expire = $expire;
    }
    public function acquire($requestId) {
        // 使用 SET key value NX PX expire - 原子操作
        $result = $this->redis->set(
            $this->lockKey, 
            $requestId, 
            ['NX', 'PX' => $this->expire * 1000]
        );
        return $result === true;
    }
    public function release($requestId) {
        // Lua 脚本保证原子性
        $script = <<<LUA
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end
LUA;
        return $this->redis->eval($script, [$this->lockKey, $requestId], 1);
    }
}
// 2. 定时任务调度器
class DistributedScheduler {
    private $redis;
    public function __construct($redis) {
        $this->redis = $redis;
    }
    /**
     * 添加延迟任务
     */
    public function addDelayTask($task, $delay, $data) {
        $id = uniqid('task_');
        $scheduleTime = time() + $delay;
        // 使用 ZSet 存储，score 为执行时间戳
        $this->redis->zAdd("delay_tasks", $scheduleTime, json_encode([
            'id' => $id,
            'task' => $task,
            'data' => $data,
            'retry_count' => 0
        ]));
        return $id;
    }
    /**
     * 执行循环 - 在每台机器上运行
     */
    public function runWorker($processLimit = 10) {
        $lock = new DistributedLock($this->redis, 'scheduler_lock');
        while (true) {
            // 获取分布式锁，防止多台机器同时消费
            $requestId = uniqid();
            if (!$lock->acquire($requestId)) {
                echo "另一台机器正在处理，等待...\n";
                sleep(1);
                continue;
            }
            try {
                // 获取到期的任务（score <= 当前时间）
                $tasks = $this->redis->zRangeByScore("delay_tasks", 0, time(), ['LIMIT' => 0, $processLimit]);
                foreach ($tasks as $taskData) {
                    $task = json_decode($taskData, true);
                    // 从 ZSet 移除（防止重复处理）
                    $removed = $this->redis->zRem("delay_tasks", $taskData);
                    if ($removed) {
                        $this->processTask($task);
                    }
                }
            } finally {
                $lock->release($requestId);
            }
            usleep(500000); // 0.5秒
        }
    }
    private function processTask($task) {
        try {
            echo "执行任务: {$task['id']} - {$task['task']}\n";
            // 这里调用实际的业务逻辑
            (new TaskHandler())->execute($task);
        } catch (Exception $e) {
            // 重试机制
            $this->retryTask($task);
        }
    }
}

优点：

实现简单,不引入额外组件
对 Redis 性能影响小

缺点：

任务持久化不足,Redis 重启数据丢失
不适合处理大量任务

方案2：基于 RabbitMQ 延迟队列（推荐使用）

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
// 1. 生产者：添加延迟任务
class TaskProducer {
    private $channel;
    public function __construct($connection) {
        $this->channel = $connection->channel();
    }
    public function addTask($taskName, $data, $delaySeconds = 0) {
        // 声明正常队列
        $this->channel->queue_declare(
            'task_queue',
            false, true, false, false
        );
        // 声明延迟交换机（使用死信交换机实现）
        $this->channel->exchange_declare(
            'delay_exchange', 
            'direct',
            false, true, false
        );
        // 声明延迟队列（带 TTL）
        $this->channel->queue_declare(
            'delay_queue',
            false, true, false, false,
            false, [
                'x-dead-letter-exchange' => ['S', 'task_exchange'],
                'x-dead-letter-routing-key' => ['S', 'task_queue'],
                'x-message-ttl' => ['I', $delaySeconds * 1000]
            ]
        );
        // 绑定延迟队列
        $this->channel->queue_bind(
            'delay_queue',
            'delay_exchange',
            'delay'
        );
        // 发送消息到延迟队列
        $message = new AMQPMessage(json_encode([
            'task' => $taskName,
            'data' => $data
        ]), [
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
            'expiration' => $delaySeconds * 1000 // 消息级 TTL
        ]);
        $this->channel->basic_publish($message, 'delay_exchange', 'delay');
        echo "任务已添加，延迟 {$delaySeconds} 秒\n";
    }
}
// 2. 消费者：执行任务
class TaskConsumer {
    private $channel;
    public function __construct($connection) {
        $this->channel = $connection->channel();
    }
    public function consume() {
        // 声明正常队列
        $this->channel->queue_declare(
            'task_queue',
            false, true, false, false
        );
        $this->channel->exchange_declare(
            'task_exchange',
            'direct',
            false, true, false
        );
        $this->channel->queue_bind(
            'task_queue',
            'task_exchange',
            'task_queue'
        );
        echo "消费者启动，等待处理消息...\n";
        $callback = function ($msg) {
            $task = json_decode($msg->body, true);
            echo "处理任务: {$task['task']}\n";
            // 执行实际业务逻辑
            $mq = MQ::getInstance();
            $mq->ack($msg->delivery_info['delivery_tag']);
        };
        $this->channel->basic_qos(null, 10, null); // 每次取10个
        $this->channel->basic_consume(
            'task_queue',
            '',
            false,
            false,
            false,
            false,
            $callback
        );
        while ($this->channel->is_consuming()) {
            $this->channel->wait();
        }
    }
}

优点：

消息持久化,保证可靠
支持消息确认、重试机制
支持集群扩展

缺点：

需要引入 RabbitMQ 组件

方案3：基于 Gearman/Job Server

// Gearman 客户端
class GearmanTask {
    private $client;
    public function __construct() {
        $this->client = new GearmanClient();
        $this->client->addServer('127.0.0.1', 4730);
    }
    // 添加延迟任务
    public function addDelayTask($taskName, $data, $delay = 0) {
        $task = new GearmanTask();
        $task->function = $taskName;
        $task->workload = json_encode($data);
        $task->unique = uniqid();
        // 添加回调
        $this->client->addTask($taskName, json_encode($data), null, $unique);
        // 使用定时器实现延迟
        if ($delay > 0) {
            $job = new GearmanJob();
            $job->runAfter = time() + $delay;
        }
        $this->client->runTasks();
    }
}
// Gearman Worker
class GearmanWorker {
    private $worker;
    public function __construct() {
        $this->worker = new GearmanWorker();
        $this->worker->addServer('127.0.0.1', 4730);
        $this->worker->addFunction('send_email', function (GearmanJob $job) {
            $data = json_decode($job->workload(), true);
            // 执行邮件发送逻辑
            echo "发送邮件到: " . $data['email'] . "\n";
        });
    }
    public function run() {
        while ($this->worker->work()) {
            if ($this->worker->returnCode() != GEARMAN_SUCCESS) {
                echo "错误: " . $this->worker->error() . "\n";
            }
        }
    }
}

方案4：使用专业调度框架（最推荐）

使用 Swoole/Workerman 内置的定时器 + Redis 分布式锁：

// 基于 Swoole 的高性能定时器
use Swoole\Timer;
use Swoole\Coroutine;
class SwooleScheduler {
    private $redis;
    public function run() {
        // 主进程运行 WebSocket 服务
        $server = new Swoole\WebSocket\Server('0.0.0.0', 9502);
        // 通过 WebSocket 发送指令控制定时任务
        $server->on('message', function ($server, $frame) {
            $cmd = json_decode($frame->data, true);
            if ($cmd['type'] == 'schedule') {
                // 添加定时任务
                Timer::tick($cmd['interval'] * 1000, function () use ($cmd) {
                    $this->executeTask($cmd['task']);
                });
            }
        });
        // 同时设置普通的定时任务
        Timer::tick(5000, function() {
            // 每5秒执行一次
            $this->checkTasks();
        });
    }
    private function executeTask($taskName) {
        // 模拟 Redis 分布式锁
        $lockKey = "lock_" . $taskName;
        $lockAcquired = $this->redis->set($lockKey, 1, ['NX', 'EX' => 60]);
        if ($lockAcquired) {
            echo "执行任务: {$taskName}\n";
            // 释放锁
            $this->redis->del($lockKey);
        }
    }
    private function checkTasks() {
        echo "检查任务...\n";
        // 从数据库或 Redis 获取任务配置
        $tasks = $this->getDueTasks();
        foreach ($tasks as $task) {
            $this->executeTask($task['name']);
        }
    }
}

实现分布式定时任务的通用模式

模式1：Leader 选举模式

// 只有一台机器（Leader）执行调度
class LeaderElection {
    private $redis;
    public function checkLeader() {
        $isLeader = $this->redis->set('cron_leader', gethostname(), ['NX', 'EX' => 30]);
        if ($isLeader) {
            echo "我是 Leader，执行调度\n";
            // 续期逻辑
            Timer::tick(20000, function() {
                $this->redis->expire('cron_leader', 30);
            });
            return true;
        }
        echo "我是从节点，不执行\n";
        return false;
    }
}

生产环境最佳实践

任务表设计

CREATE TABLE `scheduled_task` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `task_name` varchar(100) NOT NULL COMMENT '任务名称',
  `task_handler` varchar(200) NOT NULL COMMENT '处理器类',
  `task_params` json DEFAULT NULL COMMENT '任务参数',
  `schedule_type` enum('ONCE','FIXED_RATE','CRON') NOT NULL,
  `schedule_date` datetime DEFAULT NULL COMMENT '单次执行时间',
  `cron_expr` varchar(50) DEFAULT NULL COMMENT 'cron表达式',
  `next_run_at` datetime NOT NULL COMMENT '下次执行时间',
  `status` enum('ENABLED','DISABLED') NOT NULL,
  `last_exec_time` datetime DEFAULT NULL,
  `last_exec_code` tinyint(4) DEFAULT NULL,
  `retry_count` int(11) DEFAULT '0',
  `created_at` datetime NOT NULL,
  `updated_at` datetime NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

完整流程

任务配置(DB) -> 调度器(Leader) -> 任务分发(Message Queue) -> 消费者(Worker) -> 结果反馈(DB)

运维注意事项

监控：监控任务执行状态、耗时、失败率
告警：任务失败自动告警
重试：失败任务自动重试，最多N次
人为干预：支持手动触发、暂停、修改定时配置

总结推荐

场景	推荐方案
中小团队	Redis + 分布式锁
需要可靠消息	RabbitMQ 延迟队列
高性能并发	Swoole + Redis
复杂调度需求	专业调度（xxl-job、GoCron 等）