Java 定時任務 Quartz （三）—— 並發

1 前言

根據 Quartz 的設計，一個 Job 可以綁定多個 Trigger，必然會遇到並發的問題。

2 並發

2.1 復現

讓我們編寫一個並發的例子：

/**
 * @author pancc
 * @version 1.0
 */
public class AcceptConcurrentDemo {

    public static void main(String[] args) throws SchedulerException, InterruptedException {
        JobDetail detail = JobBuilder.newJob(AcceptConcurrentJob.class)
                .withIdentity("detail", "group0")
                .build();


        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity("ben_trigger")
                .usingJobData("name", "ben")
                .startNow()
                .build();

        Trigger triggers = TriggerBuilder.newTrigger()
                .withIdentity("mike_trigger")
                .usingJobData("name", "mike")
                .forJob("detail", "group0")
                .startNow()
                .build();


        Scheduler scheduler = new StdSchedulerFactory().getScheduler();

        scheduler.start();
        scheduler.scheduleJob(detail, trigger);
        scheduler.scheduleJob(triggers);
        /*
         * 6 秒鍾后關閉
         */
        Thread.sleep(6_000);
        scheduler.shutdown();
    }

    @Data
    public static class AcceptConcurrentJob implements Job {
        private String name;

        @Override
        public void execute(JobExecutionContext context) {
            try {
                System.out.printf("i am %s \n", name);
                Thread.sleep(2_000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

 

請注意上邊的 Details 的 Identity ，設置為 group0.detail，同時我們創建了兩個 Trigger，第二個 trigger 在創建的時候通過指定 Identity 綁定到了目標 Job，接着提交這個 Job，與兩個 Trigger ，可以看到兩個觸發器同時出發了 Job 的 execute 方法

 

上邊的代碼也可以簡化為以下形式：

/**
 * @author pancc
 * @version 1.0
 */
public class AcceptConcurrentDemo {

    public static void main(String[] args) throws SchedulerException, InterruptedException {
        JobDetail detail = JobBuilder.newJob(AcceptConcurrentJob.class)
                .withIdentity("detail", "group0")
                .build();


        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity("ben_trigger")
                .usingJobData("name", "ben")
                .startNow()
                .build();

        Trigger triggers = TriggerBuilder.newTrigger()
                .withIdentity("mike_trigger")
                .usingJobData("name", "mike")
                .startNow()
                .build();


        Scheduler scheduler = new StdSchedulerFactory().getScheduler();

        scheduler.start();
        scheduler.scheduleJob(detail, Sets.newHashSet(trigger,triggers),true);
        /*
         * 6 秒鍾后關閉
         */
        Thread.sleep(6_000);
        scheduler.shutdown();
    }

    @Data
    public static class AcceptConcurrentJob implements Job {
        private String name;

        @Override
        public void execute(JobExecutionContext context) {
            try {
                System.out.printf("i am %s \n", name);
                Thread.sleep(2_000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

 

2.2 避免並發

為了避免並發，我們可以使用官方提供的注解 @DisallowConcurrentExecution，通過在 類上增加這個注解，我們可以觀察到第二個 trigger 進行了排隊處理：

/**
 * @author pancc
 * @version 1.0
 */
public class RejectConcurrentDemo {

    public static void main(String[] args) throws SchedulerException, InterruptedException {
        JobDetail detail = JobBuilder.newJob(RejectConcurrentJob.class)
                .withIdentity("detail", "group0")
                .build();


        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity("ben_trigger")
                .usingJobData("name", "ben")
                .startNow()
                .build();

        Trigger triggers = TriggerBuilder.newTrigger()
                .withIdentity("mike_trigger")
                .usingJobData("name", "mike")
                .forJob("detail", "group0")
                .startNow()
                .build();


        Scheduler scheduler = new StdSchedulerFactory().getScheduler();

        scheduler.start();
        scheduler.scheduleJob(detail, trigger);
        scheduler.scheduleJob(triggers);
        /*
         * 6 秒鍾后關閉
         */
        Thread.sleep(6_000);
        scheduler.shutdown();
    }


    @DisallowConcurrentExecution
    @Data
    public static class RejectConcurrentJob implements Job {
        private String name;

        @Override
        public void execute(JobExecutionContext context) {
            try {
                System.out.printf("i am %s \n", name);
                Thread.sleep(2_000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

 

 

3 避免並發的原理探索

讓我們找到 JobStore 的實現類，在這里是 RAMJobStore，點進去方法 org.quartz.simpl.RAMJobStore#acquireNextTriggers，可以看到這個方法的某個塊：

 

 通過對 Job 類上的是否存在  DisallowConcurrentExecution 注解，如果存在，表示拒絕並發執行 execute 方法。如果與即將執行的 Trigger 調用同一個 JobDetail 對象，則將當前 trigger 放入等待列表。

 

之后當前一個 Trigger 執行完畢，將等待中的 Trigger 重新加回去 RAMJobStore 持有的 trigger 列表中，等待下一次調用發生。