任务调度

package com.enjoy.schedule;

import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 2021-12-17
 * 单线程调度时（没有开启 ScheduleConfig 配置），getTask1和getTask2是同一线程调度，也就是同步执行
 *
 */
@Component
@EnableScheduling
public class TaskConfig {

    private static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-mm-dd HH:mm:ss");

    @Scheduled(fixedDelayString = "5000") //单机
    public void getTask1() {
        //竞争锁逻辑代码 .....
        System.out.println("getTask1，当前时间：" + dateFormat.format(new Date())+",线程号："+Thread.currentThread().getName());
        // throw new RuntimeException("xxxxx");
    }

    @Scheduled(cron = "0/5 * *  * * ?")
    public void getTask2() {
        System.out.println("getTask2，当前时间：" + dateFormat.format(new Date())+",线程号："+Thread.currentThread().getName());
    }

}

package com.enjoy.schedule;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * 2021-12-17
 * 没有此配置，则schedule为单线程串行执行
 */
@Configuration
@EnableScheduling
public class ScheduleConfig implements SchedulingConfigurer {

    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        taskRegistrar.setScheduler(taskExecutor());
    }

    /**
     * 配置线程池---触发器和任务共用的线程池
     * @return
     */
    @Bean(destroyMethod="shutdown")
    public Executor taskExecutor() {
        return Executors.newScheduledThreadPool(10);
    }
}

package com.hong.job;

import com.cxytiandi.elasticjob.annotation.ElasticJobConf;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.hong.business.EnjoyBusiness;
import org.springframework.beans.factory.annotation.Autowired;

@ElasticJobConf(
    name = "自定义EjoySimpleJob",//自定义
    cron = "0/5 * * * * ?",
    shardingItemParameters = "0=beijing|shenzhen,1=shanghai", //数据分片参数，固定格式 0=param0,1=param1,2=param21|param22
    shardingTotalCount = 2,//上面数据切片的数量
    listener = "com.hong.handle.MessageElasticJobListener",//任务监听
    jobExceptionHandler = "com.hong.handle.CustomJobExceptionHandler"//自定义异常处理
)
public class EnjoySimpleJob implements SimpleJob {
    @Autowired
    private EnjoyBusiness enjoyBusiness;

    /**
     * 启动后根据设定的cron定时去调用这个方法
     * @param context
     */
    public void execute(ShardingContext context) {
        System.out.println("EnjoySimpleJob,当前分片："+context.getShardingParameter());

        //当前起始
        //context.getShardingParameter(),回返切片信息beijing
        String sql = enjoyBusiness.getSql(context.getShardingParameter());
        enjoyBusiness.process(sql);
    }

}

package com.hong.handle;

import com.dangdang.ddframe.job.executor.ShardingContexts;
import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 作业监听器
 */
public class MessageElasticJobListener implements ElasticJobListener {

    @Override
    public void beforeJobExecuted(ShardingContexts shardingContexts) {
        String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        String msg = date + " 【任务开始执行-" + shardingContexts.getJobName() + "】";
        System.out.println("beforeJobExecuted执行前给管理发邮件："+msg);

    }

    @Override
    public void afterJobExecuted(ShardingContexts shardingContexts) {
        String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        String msg = date + " 【任务执行结束-" + shardingContexts.getJobName() + "】" ;
        System.out.println("afterJobExecuted执行后给管理发邮件："+msg);
        System.out.println();
    }

}

package com.hong.handle;

import com.dangdang.ddframe.job.executor.handler.JobExceptionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 自定义异常处理
 */
public class CustomJobExceptionHandler implements JobExceptionHandler {

    private Logger logger = LoggerFactory.getLogger(CustomJobExceptionHandler.class);

    @Override
    public void handleException(String jobName, Throwable cause) {
        logger.error(String.format("Job '%s' exception occur in job processing", jobName), cause);
        System.out.println("自定义作业异常处理=========》给管理发邮件：【"+jobName+"】任务异常。" + cause.getMessage());
    }

}