hyperf框架组件crontab分析

// Hyperf\Crontab\Listener\CrontabRegisterListener::class
// 监听process进程事件
public function listen(): array
{
    return [
        BeforeProcessHandle::class,
        BeforeCoroutineHandle::class,
    ];
}

// 解析crontab配置
public function process(object $event)
{
    $crontabs = $this->parseCrontabs();
    foreach ($crontabs as $crontab) {
        if ($crontab instanceof Crontab) {
            // 验证配置及是否开启crontab任务执行, crontab.enable状态控制
            $this->crontabManager->register($crontab);
        }
    }
}

// Hyperf\Crontab\Process\CrontabDispatcherProcess::class
public function handle(): void
{
    // 这个事件貌似没实现具体逻辑
    $this->event->dispatch(new CrontabDispatcherStarted());
    // 死循环
    while (ProcessManager::isRunning()) {
        // 按照 秒 粒度执行，所以crontab组件执行秒级别定时任务
        $this->sleep();
        // 解析crontab配置 获取需要执行的任务
        $crontabs = $this->scheduler->schedule();
        while (! $crontabs->isEmpty()) {
            $crontab = $crontabs->dequeue();
            // 根据策略调度不同类型worker, 默认Hyperf\Crontab\Strategy\WorkerStrategy, 可通过配置重写该类
            $this->strategy->dispatch($crontab);
        }
    }
}

private function sleep()
{
    $current = date('s', time());
    $sleep = 60 - $current;
    $this->logger->debug('Crontab dispatcher sleep ' . $sleep . 's.');
    $sleep > 0 && \Swoole\Coroutine::sleep($sleep);
}

// Hyperf\Crontab\Strategy\WorkerStrategy::class
// 调度逻辑
public function dispatch(Crontab $crontab)
{
    $server = $this->serverFactory->getServer()->getServer();
    if ($server instanceof Server && $crontab->getExecuteTime() instanceof Carbon) {
        $workerId = $this->getNextWorkerId($server);
        // 任务推送到Pipe中，Hyperf\Crontab\Listener\OnPipeMessageListener::class监听器处理
        $server->sendMessage(new PipeMessage(
            'callback',
            [Executor::class, 'execute'],
            $crontab
        ), $workerId);
    }
}

// Hyperf\Crontab\Strategy\Executor::class负责执行不同策略
public function execute(Crontab $crontab)
{
    // ...
    // 默认采用callback
    $diff = $crontab->getExecuteTime()->diffInRealSeconds(new Carbon());
    $callback = null;
    switch ($crontab->getType()) {
        case 'callback':
            [$class, $method] = $crontab->getCallback();
            $parameters = $crontab->getCallback()[2] ?? null;
            if ($class && $method && class_exists($class) && method_exists($class, $method)) {
                $callback = function () use ($class, $method, $parameters, $crontab) {
                    // ...
                    // 根据配置不同执行互斥锁，默认是直接执行，不区分多实例场景
                    Coroutine::create($this->decorateRunnable($crontab, $runnable));
                };
            }
            break;
            // ...
    }
}

// 配置策略
protected function decorateRunnable(Crontab $crontab, Closure $runnable): Closure
{
    if ($crontab->isSingleton()) {
        $runnable = $this->runInSingleton($crontab, $runnable);
    }

    if ($crontab->isOnOneServer()) {
        $runnable = $this->runOnOneServer($crontab, $runnable);
    }

    return $runnable;
}

protected function runInSingleton(Crontab $crontab, Closure $runnable): Closure
{
    return function () use ($crontab, $runnable) {
        $taskMutex = $this->getTaskMutex();
        // 判断redis是否创建锁成功，采用`setnx`指令
        if ($taskMutex->exists($crontab) || ! $taskMutex->create($crontab)) {
            $this->logger->info(sprintf('Crontab task [%s] skipped execution at %s.', $crontab->getName(), date('Y-m-d H:i:s')));
            return;
        }
        // 加锁->任务->删锁
        // 这种类型存在缺陷，如果实例进程被Kill掉，无法执行 删除锁操作，导致其他实例无法获取锁
        // 锁的默认时间：1h
        try {
            $runnable();
        } finally {
            $taskMutex->remove($crontab);
        }
    };
}

protected function runOnOneServer(Crontab $crontab, Closure $runnable): Closure
{
    return function () use ($crontab, $runnable) {
        $taskMutex = $this->getServerMutex();
        // 尝试获取锁，失败则跳过
        if (! $taskMutex->attempt($crontab)) {
            $this->logger->info(sprintf('Crontab task [%s] skipped execution at %s.', $crontab->getName(), date('Y-m-d H:i:s')));
            return;
        }

        $runnable();
    };
}

// Hyperf\Crontab\Mutex\RedisServerMutex::class
public function attempt(Crontab $crontab): bool
{
    // ...
    $redis = $this->redisFactory->get($crontab->getMutexPool());
    $mutexName = $this->getMutexName($crontab);
    // 尝试加锁
    $result = (bool) $redis->set($mutexName, $this->macAddress, ['NX', 'EX' => $crontab->getMutexExpires()]);

    if ($result === true) {
        // 启动coroutine监听`SIGTERM`信号，程序退出主动删除 锁
        // 在容器场景这里会出现坑
        Coroutine::create(function () use ($crontab, $redis, $mutexName) {
            $exited = CoordinatorManager::until(Constants::WORKER_EXIT)->yield($crontab->getMutexExpires());
            $exited && $redis->del($mutexName);
        });
        return $result;
    }

}

public function dispatch(Crontab $crontab)
{
    $server = $this->serverFactory->getServer()->getServer();
    if ($server instanceof Server && $crontab->getExecuteTime() instanceof Carbon) {
        // ...
        // 监听,主动删除
        pcntl_signal(SIGTERM, function () use ($crontab){
            $taskMutex = $this->container->get(TaskMutex::class);
            if ($taskMutex instanceof TaskMutex) {
                $taskMutex->remove($crontab);
            }
        });
    }
}