IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

PHP定时任务分布式锁

wen PHP项目 4

本文目录导读：

PHP定时任务分布式锁

  1. 为什么需要分布式锁
  2. 主流实现方案
  3. 完整的生产级实现
  4. 最佳实践建议
  5. 注意事项

我来为您详细讲解PHP定时任务的分布式锁实现方案。

为什么需要分布式锁

在分布式环境中,多个服务器节点可能同时执行同一个定时任务，导致：

  • 数据重复处理
  • 资源竞争
  • 任务重复执行

主流实现方案

Redis分布式锁（最常用）

<?php
class RedisDistributedLock
{
    private $redis;
    private $lockKey;
    private $lockValue;
    private $expireTime;
    public function __construct($redis, $lockKey, $expireTime = 10)
    {
        $this->redis = $redis;
        $this->lockKey = $lockKey;
        $this->expireTime = $expireTime;
        $this->lockValue = uniqid('lock_', true) . '_' . mt_rand();
    }
    /**
     * 获取锁（非阻塞）
     */
    public function acquire()
    {
        // 使用SET NX EX原子操作
        $result = $this->redis->set(
            $this->lockKey,
            $this->lockValue,
            ['NX', 'EX' => $this->expireTime]
        );
        return $result === true;
    }
    /**
     * 获取锁（阻塞重试）
     */
    public function acquireWithRetry($maxRetry = 10, $retryDelay = 0.2)
    {
        $attempt = 0;
        while ($attempt < $maxRetry) {
            if ($this->acquire()) {
                return true;
            }
            $attempt++;
            usleep($retryDelay * 1000000);
        }
        return false;
    }
    /**
     * 释放锁（使用Lua脚本保证原子性）
     */
    public function release()
    {
        $luaScript = <<<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(
            $luaScript,
            [$this->lockKey, $this->lockValue],
            1
        ) == 1;
    }
    /**
     * 获取锁值（用于调试）
     */
    public function getLockValue()
    {
        return $this->lockValue;
    }
}
// 使用示例
class CronTask
{
    private $lock;
    public function execute()
    {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        $this->lock = new RedisDistributedLock(
            $redis,
            'cron:task:sync_data',
            60 // 锁超时时间
        );
        if (!$this->lock->acquire()) {
            echo "任务正在其他节点执行，跳过...\n";
            return false;
        }
        try {
            // 执行具体任务
            $this->doTask();
        } finally {
            // 释放锁
            $this->lock->release();
        }
        return true;
    }
    private function doTask()
    {
        // 具体的业务逻辑
        echo "执行同步数据任务...\n";
        sleep(5);
    }
}

使用Redlock算法（高可靠性）

<?php
class RedLock
{
    private $redisNodes = [];
    private $quorum;
    private $retryCount = 3;
    private $retryDelay = 200; // 毫秒
    public function __construct(array $redisNodes)
    {
        $this->redisNodes = $redisNodes;
        $this->quorum = intval(count($redisNodes) / 2) + 1;
    }
    /**
     * 获取锁
     */
    public function lock($resource, $ttl)
    {
        $token = uniqid('redlock_', true);
        $startTime = microtime(true) * 1000;
        $successCount = 0;
        $locks = [];
        // 尝试在所有节点获取锁
        foreach ($this->redisNodes as $node) {
            $result = $this->tryLockNode($node, $resource, $token, $ttl);
            if ($result) {
                $successCount++;
                $locks[] = $node;
            }
        }
        // 计算总耗时
        $elapsedTime = microtime(true) * 1000 - $startTime;
        // 检查是否获取到大多数节点的锁且耗时小于TTL
        if ($successCount >= $this->quorum && $elapsedTime < $ttl) {
            $this->locks = $locks;
            $this->resource = $resource;
            $this->token = $token;
            return true;
        }
        // 获取失败，释放已获取的锁
        $this->unlock($resource, $token);
        return false;
    }
    /**
     * 释放锁
     */
    public function unlock($resource, $token)
    {
        foreach ($this->redisNodes as $node) {
            $this->unlockNode($node, $resource, $token);
        }
    }
    private function tryLockNode($node, $resource, $token, $ttl)
    {
        try {
            $redis = new Redis();
            $redis->connect($node['host'], $node['port']);
            return $redis->set(
                $resource,
                $token,
                ['NX', 'PX' => $ttl]
            );
        } catch (Exception $e) {
            return false;
        }
    }
    private function unlockNode($node, $resource, $token)
    {
        $luaScript = <<<LUA
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end
LUA;
        try {
            $redis = new Redis();
            $redis->connect($node['host'], $node['port']);
            $redis->eval($luaScript, [$resource, $token], 1);
        } catch (Exception $e) {
            // 记录日志
        }
    }
}

数据库分布式锁

<?php
class DatabaseDistributedLock
{
    private $pdo;
    private $lockTable = 'cron_locks';
    public function __construct(PDO $pdo)
    {
        $this->pdo = $pdo;
        $this->initTable();
    }
    /**
     * 初始化锁表
     */
    private function initTable()
    {
        $sql = "CREATE TABLE IF NOT EXISTS {$this->lockTable} (
            lock_key VARCHAR(255) PRIMARY KEY,
            lock_token VARCHAR(255) NOT NULL,
            expires_at DATETIME NOT NULL,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )";
        $this->pdo->exec($sql);
    }
    /**
     * 获取锁
     */
    public function acquire($lockKey, $expireSeconds = 60)
    {
        $token = uniqid('db_lock_', true);
        $expiresAt = date('Y-m-d H:i:s', time() + $expireSeconds);
        try {
            // 尝试插入锁记录
            $stmt = $this->pdo->prepare(
                "INSERT INTO {$this->lockTable} (lock_key, lock_token, expires_at) 
                 VALUES (:key, :token, :expires)"
            );
            $stmt->execute([
                ':key' => $lockKey,
                ':token' => $token,
                ':expires' => $expiresAt
            ]);
            return $token;
        } catch (PDOException $e) {
            // 锁已存在，检查是否过期
            if ($e->getCode() == '23000') { // 唯一约束冲突
                return $this->handleExistingLock($lockKey, $expireSeconds);
            }
            throw $e;
        }
    }
    /**
     * 处理已存在的锁
     */
    private function handleExistingLock($lockKey, $expireSeconds)
    {
        // 尝试删除过期锁
        $stmt = $this->pdo->prepare(
            "DELETE FROM {$this->lockTable} 
             WHERE lock_key = :key AND expires_at < NOW()"
        );
        $stmt->execute([':key' => $lockKey]);
        // 再次尝试获取锁
        return $this->acquire($lockKey, $expireSeconds);
    }
    /**
     * 释放锁
     */
    public function release($lockKey, $token)
    {
        $stmt = $this->pdo->prepare(
            "DELETE FROM {$this->lockTable} 
             WHERE lock_key = :key AND lock_token = :token"
        );
        $stmt->execute([
            ':key' => $lockKey,
            ':token' => $token
        ]);
        return $stmt->rowCount() > 0;
    }
}

基于文件锁（单机方案）

<?php
class FileDistributedLock
{
    private $lockDir;
    private $lockFile;
    public function __construct($lockName, $lockDir = '/tmp/cron_locks')
    {
        $this->lockDir = $lockDir;
        $this->lockFile = $lockDir . '/' . md5($lockName) . '.lock';
        if (!is_dir($lockDir)) {
            mkdir($lockDir, 0777, true);
        }
    }
    /**
     * 获取锁
     */
    public function acquire($timeout = 10)
    {
        $startTime = time();
        while (true) {
            // 尝试创建锁文件
            $handle = @fopen($this->lockFile, 'x');
            if ($handle) {
                fwrite($handle, json_encode([
                    'pid' => getmypid(),
                    'time' => date('Y-m-d H:i:s'),
                    'host' => gethostname()
                ]));
                fclose($handle);
                return true;
            }
            // 检查锁是否过期
            if (file_exists($this->lockFile)) {
                $fileTime = filemtime($this->lockFile);
                if (time() - $fileTime > $timeout) {
                    // 锁过期，强制移除
                    @unlink($this->lockFile);
                    continue;
                }
            }
            // 等待重试
            if (time() - $startTime > $timeout) {
                return false;
            }
            usleep(500000); // 0.5秒
        }
    }
    /**
     * 释放锁
     */
    public function release()
    {
        if (file_exists($this->lockFile)) {
            @unlink($this->lockFile);
            return true;
        }
        return false;
    }
}

完整的生产级实现

<?php
/**
 * 生产级定时任务管理器
 */
class CronTaskManager
{
    private $lock;
    private $taskName;
    private $config;
    private $logger;
    public function __construct($taskName, $config = [])
    {
        $this->taskName = $taskName;
        $this->config = array_merge([
            'lock_type' => 'redis',        // redis/database/file
            'lock_expire' => 300,           // 锁过期时间(秒)
            'redis' => [
                'host' => '127.0.0.1',
                'port' => 6379,
                'password' => null,
                'db' => 0
            ],
            'database' => [
                'dsn' => 'mysql:host=localhost;dbname=test',
                'username' => 'root',
                'password' => 'root'
            ],
            'log_path' => '/var/log/cron_tasks/'
        ], $config);
        $this->initLock();
        $this->initLogger();
    }
    /**
     * 初始化锁
     */
    private function initLock()
    {
        $lockKey = "cron:task:{$this->taskName}";
        switch ($this->config['lock_type']) {
            case 'redis':
                $redis = new Redis();
                $redis->connect(
                    $this->config['redis']['host'],
                    $this->config['redis']['port']
                );
                if ($this->config['redis']['password']) {
                    $redis->auth($this->config['redis']['password']);
                }
                $redis->select($this->config['redis']['db']);
                $this->lock = new RedisDistributedLock(
                    $redis,
                    $lockKey,
                    $this->config['lock_expire']
                );
                break;
            case 'database':
                $pdo = new PDO(
                    $this->config['database']['dsn'],
                    $this->config['database']['username'],
                    $this->config['database']['password']
                );
                $this->lock = new DatabaseDistributedLock($pdo);
                break;
            case 'file':
                $this->lock = new FileDistributedLock(
                    $this->taskName,
                    $this->config['log_path'] . 'locks/'
                );
                break;
        }
    }
    /**
     * 初始化日志
     */
    private function initLogger()
    {
        $logDir = $this->config['log_path'] . date('Y/m');
        if (!is_dir($logDir)) {
            mkdir($logDir, 0777, true);
        }
        $this->logger = $logDir . '/' . date('d') . '.log';
    }
    /**
     * 记录日志
     */
    private function log($message, $type = 'INFO')
    {
        $logMessage = sprintf(
            "[%s] [%s] [%s] %s\n",
            date('Y-m-d H:i:s'),
            $type,
            $this->taskName,
            $message
        );
        error_log($logMessage, 3, $this->logger);
    }
    /**
     * 执行任务
     */
    public function run(callable $task)
    {
        if (!$this->acquireLock()) {
            $this->log("任务被拒绝，已有其他节点在执行");
            return false;
        }
        $startTime = microtime(true);
        $this->log("任务开始执行");
        try {
            // 执行任务
            $result = $task();
            $execTime = round(microtime(true) - $startTime, 3);
            $this->log("任务执行成功，耗时: {$execTime}秒", 'SUCCESS');
            return $result;
        } catch (Exception $e) {
            $this->log("任务执行失败: " . $e->getMessage() . 
                       "\n" . $e->getTraceAsString(), 'ERROR');
            return false;
        } finally {
            // 确保释放锁
            $this->releaseLock();
        }
    }
    /**
     * 获取锁
     */
    private function acquireLock()
    {
        try {
            if ($this->lock instanceof RedisDistributedLock) {
                return $this->lock->acquireWithRetry(5, 0.2);
            } elseif ($this->lock instanceof DatabaseDistributedLock) {
                return $this->lock->acquire(
                    "cron:task:{$this->taskName}",
                    $this->config['lock_expire']
                ) ? true : false;
            } elseif ($this->lock instanceof FileDistributedLock) {
                return $this->lock->acquire($this->config['lock_expire']);
            }
            return false;
        } catch (Exception $e) {
            $this->log("获取锁失败: " . $e->getMessage(), 'ERROR');
            return false;
        }
    }
    /**
     * 释放锁
     */
    private function releaseLock()
    {
        try {
            if ($this->lock instanceof RedisDistributedLock) {
                return $this->lock->release();
            } elseif ($this->lock instanceof DatabaseDistributedLock) {
                return $this->lock->release(
                    "cron:task:{$this->taskName}",
                    $this->lock->getLockToken()
                );
            } elseif ($this->lock instanceof FileDistributedLock) {
                return $this->lock->release();
            }
        } catch (Exception $e) {
            $this->log("释放锁失败: " . $e->getMessage(), 'ERROR');
        }
        return false;
    }
}
// 使用示例
$taskManager = new CronTaskManager('sync_user_data', [
    'lock_type' => 'redis',
    'lock_expire' => 300,
    'redis' => [
        'host' => '127.0.0.1',
        'port' => 6379
    ]
]);
$result = $taskManager->run(function() {
    // 执行具体的定时任务逻辑
    $userService = new UserService();
    return $userService->syncUsers();
});

最佳实践建议

锁的续期处理

class RenewLockWorker
{
    private $redis;
    private $lockKey;
    private $lockValue;
    private $ttl;
    private $renewInterval;
    public function start($lockKey, $lockValue, $ttl = 60, $renewInterval = 30)
    {
        // 启动守护线程定时续期
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
        $this->lockKey = $lockKey;
        $this->lockValue = $lockValue;
        $this->ttl = $ttl;
        $this->renewInterval = $renewInterval;
        $this->renewLoop();
    }
    private function renewLoop()
    {
        while (true) {
            usleep($this->renewInterval * 1000000);
            $luaScript = <<<LUA
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("pexpire", KEYS[1], ARGV[2])
else
    return 0
end
LUA;
            $this->redis->eval(
                $luaScript,
                [$this->lockKey, $this->lockValue, $this->ttl * 1000],
                1
            );
            // 检查是否需要停止
            if (pcntl_signal_dispatch()) {
                break;
            }
        }
    }
}

锁的监控和告警

class LockMonitor
{
    private $redis;
    public function checkLockHealth($lockKey)
    {
        $lockInfo = $this->redis->get($lockKey);
        if (!$lockInfo) {
            $this->sendAlert("锁不存在: {$lockKey}");
            return false;
        }
        $ttl = $this->redis->ttl($lockKey);
        if ($ttl < 10) {
            $this->sendAlert("锁即将过期: {$lockKey}, TTL: {$ttl}秒");
        }
        return true;
    }
    private function sendAlert($message)
    {
        // 发送邮件、短信或企业微信通知
        $alertService = new AlertService();
        $alertService->send($message);
    }
}

任务执行记录

class TaskExecutionLog
{
    private $pdo;
    public function logStart($taskName, $host, $pid)
    {
        $stmt = $this->pdo->prepare(
            "INSERT INTO task_executions 
             (task_name, host, pid, status, start_time) 
             VALUES (?, ?, ?, 'running', NOW())"
        );
        $stmt->execute([$taskName, $host, $pid]);
        return $this->pdo->lastInsertId();
    }
    public function logEnd($executionId, $status, $message = '')
    {
        $stmt = $this->pdo->prepare(
            "UPDATE task_executions 
             SET status = ?, message = ?, end_time = NOW() 
             WHERE id = ?"
        );
        return $stmt->execute([$status, $message, $executionId]);
    }
}

注意事项

  1. 锁超时设置：根据任务执行时间合理设置，避免任务未完成锁就过期
  2. 原子性保证：获取锁和释放锁的操作必须是原子性的
  3. 异常处理：确保在任务异常时也能正确释放锁
  4. 监控告警：及时发现锁的异常情况
  5. 兼容性：考虑各种分布式环境的兼容性

这套方案可以根据具体需求选择合适的实现方式,建议生产环境使用Redis方案，并配合完善的任务监控机制。

上一篇PHP定时扫描到期订单

下一篇PHP怎么保证消息不丢

相关文章

抱歉，评论功能暂时关闭!