Hyperf框架实现动态配置定时任务的技术方案

2025-06-02 03:13:25作者：董宙帆

概述

在Hyperf框架中实现动态配置定时任务是一个常见的业务需求，特别是在需要灵活调整任务执行策略的场景下。本文将详细介绍如何在Hyperf中通过数据库动态管理定时任务，包括路由规则、执行规则以及任务状态的配置。

核心实现思路

1. 数据库设计

首先需要设计一个数据库表来存储定时任务的配置信息，典型的字段包括：

任务ID
任务名称
路由规则（执行哪个控制器方法）
Cron表达式（执行时间规则）
任务状态（启用/禁用）
任务参数
创建时间
更新时间

2. 管理界面实现

创建一个管理界面用于CRUD操作定时任务配置，这通常包括：

任务列表展示
新增任务表单
编辑任务表单
启用/禁用操作
立即执行操作

3. 动态任务调度机制

实现动态任务调度的核心流程：

当检测到任务配置变更时（通过事务完成确认）
优雅地关闭当前运行的Cron Worker（通过事件通知）
从数据库读取最新的任务配置
根据新配置重新初始化Cron Worker
启动新的Cron Worker（通过事件通知）

具体实现方案

数据库配置模型

<?php

namespace App\Model;

use Hyperf\DbConnection\Model\Model;

class CronJob extends Model
{
    protected $table = 'cron_jobs';
    
    protected $fillable = [
        'name',
        'route',
        'rule',
        'status',
        'params',
    ];
    
    protected $casts = [
        'params' => 'array',
        'status' => 'boolean',
    ];
}

任务配置服务

<?php

namespace App\Service;

use App\Model\CronJob;
use Hyperf\Crontab\Crontab;

class CronJobService
{
    public function getEnabledJobs(): array
    {
        return CronJob::where('status', true)->get()->map(function ($job) {
            $crontab = new Crontab();
            $crontab->setName($job->name);
            $crontab->setRule($job->rule);
            $crontab->setCallback([$job->route, 'execute']);
            $crontab->setMemo($job->name);
            $crontab->setEnable(true);
            return $crontab;
        })->toArray();
    }
}

任务调度器

<?php

namespace App\Process;

use App\Service\CronJobService;
use Hyperf\Crontab\Strategy\Executor;
use Hyperf\Process\AbstractProcess;
use Hyperf\Process\Annotation\Process;
use Psr\Container\ContainerInterface;

#[Process(name: "dynamic-cron")]
class DynamicCronProcess extends AbstractProcess
{
    protected $server;
    protected $executor;
    protected $jobs = [];
    
    public function __construct(ContainerInterface $container)
    {
        parent::__construct($container);
        $this->executor = $container->get(Executor::class);
    }
    
    public function handle(): void
    {
        $service = $this->container->get(CronJobService::class);
        $this->jobs = $service->getEnabledJobs();
        
        while (true) {
            $this->checkAndRun();
            sleep(1);
        }
    }
    
    protected function checkAndRun(): void
    {
        $now = time();
        foreach ($this->jobs as $job) {
            if ($job->isDue($now)) {
                $this->executor->execute($job);
            }
        }
    }
    
    public function reloadJobs(): void
    {
        $service = $this->container->get(CronJobService::class);
        $this->jobs = $service->getEnabledJobs();
    }
}

配置变更监听

<?php

namespace App\Listener;

use App\Process\DynamicCronProcess;
use Hyperf\Event\Annotation\Listener;
use Hyperf\Event\Contract\ListenerInterface;
use Hyperf\Framework\Event\OnPipeMessage;

#[Listener]
class CronConfigChangeListener implements ListenerInterface
{
    public function listen(): array
    {
        return [
            OnPipeMessage::class,
        ];
    }
    
    public function process(object $event): void
    {
        if ($event instanceof OnPipeMessage && $event->data['type'] === 'reload_cron') {
            $process = $event->data['process'];
            if ($process instanceof DynamicCronProcess) {
                $process->reloadJobs();
            }
        }
    }
}