elastic-job动态添加定时任务

时间：2022-12-25 19:01:25

在elastic-job的使用过程中，我们会遇到动态添加定时任务的时候，但是官网上面并没有对这块内容进行说明。按照我的理解以及官网上面elastic-job的框架图，ej的定时任务其实是存储在zookeeper的一个个节点上面，所以通过给zookeeper添加对应的节点即可完成定时任务的添加动作。

下面上代码：

import java.text.SimpleDateFormat;

import java.util.Date;

import com.dangdang.ddframe.job.api.ShardingContext;

import com.dangdang.ddframe.job.api.simple.SimpleJob;

import com.dangdang.ddframe.job.config.JobCoreConfiguration;

import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;

import com.dangdang.ddframe.job.exception.JobSystemException;

import com.dangdang.ddframe.job.lite.api.JobScheduler;

import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;

import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;

import org.springframework.context.ApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

public class DynamicAddJob implements SimpleJob{

	private static final String CRON_DATE_FORMAT = "ss mm HH dd MM ? yyyy";

	/***

	 * @param date 时间

	 * @return cron类型的日期

	 */

	public static String getCron(final Date date) {

		SimpleDateFormat sdf = new SimpleDateFormat(CRON_DATE_FORMAT);

		String formatTimeStr = "";

		if (date != null) {

			formatTimeStr = sdf.format(date);

		}

		return formatTimeStr;

	}

	public static void main(String[] args){

		ApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring/applicationContext-job.xml");

		ZookeeperRegistryCenter zookeeperRegistryCenter = context.getBean(ZookeeperRegistryCenter.class);

		long now = System.currentTimeMillis();

		for (int i = 0; i < 100; i++) {

			String cron = getCron(new Date(now + (i + 1) * 50000));

			JobCoreConfiguration coreConfig = JobCoreConfiguration.newBuilder("dynamicDemoJob-" + i, cron, 2).build();

			SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(coreConfig, DynamicAddJob.class.getCanonicalName());

			JobScheduler jobScheduler = new JobScheduler(zookeeperRegistryCenter, LiteJobConfiguration.newBuilder(simpleJobConfig).build());

			try {

				jobScheduler.init();

			}catch (JobSystemException e){

				e.printStackTrace();

			}

		}

	}

	@Override

	public void execute(ShardingContext shardingContext) {

		switch (shardingContext.getShardingItem()){

			case 0:

				System.out.println("doing sharding 0...job name is "+shardingContext.getJobName());

				// do something by sharding item 0

				break;

			case 1:

				System.out.println("doing sharding 1...job name is "+shardingContext.getJobName());

				// do something by sharding item 1

				break;

		}

	}

}

这里用到比较重要的一个类是JobScheduler，这是lite-core里面一个比较核心的类，这个类其实就是我们的job，他的构造方法包含以下参数：

  • CoordinatorRegistryCenter regCenter：注册中心，这里是zookeeper
  • LiteJobConfiguration liteJobConfig：定时任务的配置信息

这里可以看一下LiteJobConfiguration这个类，采用了设计模式中的建造者模式进行构建。可能看着会比较摸不着头脑，里面的Builder跟平时的不太一样，这里我们需要知道的是ej的源码采用了lombok这个代码简化的工具，只需要通过注解的形式就能将我们平时所需要的get/set和构造器的内容在编译时创建出来，不需要在代码中体现，能够大大简化我们的代码。

另外还遇到一个坑。这段代码不能重复使用，第一次跑的时候没问题，过段时间再次跑这个代码时，会在init()处报错，原因是我们新建的job根本不能被fire，我跟了进去。发现，job的cron表达式表示的时间还是以前的时间，这就奇怪了，明明我这边配置了一个新的时间。通过debug，进入init方法中，发现他会更新job信息，而更新时，会去zk上面load配置信息，而zk的znode节点是老的节点，上面存储的配置信息也是老的，所以这块的cron表达式也是旧的时间，根本不会被执行，下面贴出源码，供大家参考。

init()源码：

    /**

     * 初始化作业.

     */

    public void init() {

        LiteJobConfiguration liteJobConfigFromRegCenter = schedulerFacade.updateJobConfiguration(liteJobConfig);

        JobRegistry.getInstance().setCurrentShardingTotalCount(liteJobConfigFromRegCenter.getJobName(), liteJobConfigFromRegCenter.getTypeConfig().getCoreConfig().getShardingTotalCount());

        JobScheduleController jobScheduleController = new JobScheduleController(createScheduler(), createJobDetail(liteJobConfigFromRegCenter.getTypeConfig().getJobClass()), liteJobConfigFromRegCenter.getJobName());

        JobRegistry.getInstance().registerJob(liteJobConfigFromRegCenter.getJobName(), jobScheduleController, regCenter);

        schedulerFacade.registerStartUpInfo(!liteJobConfigFromRegCenter.isDisabled());

        jobScheduleController.scheduleJob(liteJobConfigFromRegCenter.getTypeConfig().getCoreConfig().getCron());

    }

updateJobConfiguration()的源码如下：

    /**

     * 更新作业配置.

     *

     * @param liteJobConfig 作业配置

     * @return 更新后的作业配置

     */

    public LiteJobConfiguration updateJobConfiguration(final LiteJobConfiguration liteJobConfig) {

        configService.persist(liteJobConfig);

        return configService.load(false);

    }

load()源码如下：

    /**

     * 读取作业配置.

     *

     * @param fromCache 是否从缓存中读取

     * @return 作业配置

     */

    public LiteJobConfiguration load(final boolean fromCache) {

        String result;

        if (fromCache) {

            result = jobNodeStorage.getJobNodeData(ConfigurationNode.ROOT);

            if (null == result) {

                result = jobNodeStorage.getJobNodeDataDirectly(ConfigurationNode.ROOT);

            }

        } else {

            result = jobNodeStorage.getJobNodeDataDirectly(ConfigurationNode.ROOT);

        }

        return LiteJobConfigurationGsonFactory.fromJson(result);

    }

可以发现这块load有两种，一种是从缓存（这里的缓存使用Map来实现的TreeCache）中获取getJobNodeData，一种是从注册中心也就是zookeeper中获取getJobNodeDataDirectly。load的时候，根据的是zk的路径，其实也就是任务的jobName，所以我们要尽量避免任务名称的重复。

相关文章