Spring任务执行和任务调度

系统管理员 2023-07-24 05:34 12阅读 0赞

### 介绍 ###

Spring框架分别通过TaskExecutor和TaskScheduler接口提供了异步执行和任务调度的抽象。 Spring还提供了那些接口的实现，这些接口在应用程序服务器环境中支持线程池或委托给CommonJ。 最终，在公共接口后面使用这些实现可以抽象化Java SE 5，Java SE 6和Java EE环境之间的差异。

Spring还具有集成类，用于支持与Timer（自1.3起成为JDK的一部分）和Quartz Scheduler（https://www.quartz-scheduler.org/）一起进行调度。 这两个调度程序都是使用FactoryBean设置的，分别带有对Timer或Trigger实例的可选引用。 此外，还提供了Quartz Scheduler和Timer的便捷类，该类允许您调用现有目标对象的方法（类似于常规的MethodInvokingFactoryBean操作）。

### Spring TaskExecutor抽象 ###

Executors 是线程池概念的JDK名称。 “executor”的命名是由于无法保证基础实现实际上是一个池； 执行程序可能是单线程的，甚至是同步的。 Spring的抽象隐藏了Java SE和Java EE环境之间的实现细节

Spring的TaskExecutor接口与java.util.concurrent.Executor接口相同。 实际上，最初，它存在的主要原因是在使用线程池时抽象出对Java 5的需求。 该接口具有单个方法execute（Runnable task），该方法根据线程池的语义和配置接受要执行的任务。

最初创建TaskExecutor的目的是为其他Spring组件提供所需的线程池抽象。 ApplicationEventMulticaster，JMS的AbstractMessageListenerContainer和Quartz集成等组件都使用TaskExecutor抽象来池化线程。 但是，如果您的bean需要线程池行为，则可以根据自己的需要使用此抽象。

#### TaskExecutor类型 ####

Spring发行版中包含许多TaskExecutor的预构建实现。 您极有可能无需实现自己的方法。 常见的即用型变体是：

*  SyncTaskExecutor此实现不会异步执行调用。 而是，每个调用都在调用线程中进行。 它主要用于不需要多线程的情况下，例如在简单的测试案例中。
 *  SimpleAsyncTaskExecutor此实现不重用任何线程，而是为每次调用启动一个新线程。 但是，它确实支持并发限制，该限制将阻止超出限制的所有调用，直到释放插槽为止。 如果您正在寻找真正的池，请参见下面的ThreadPoolTaskExecutor。
 *  ConcurrentTaskExecutor此实现是java.util.concurrent.Executor实例的适配器。 还有一个替代方法ThreadPoolTaskExecutor，它将Executor配置参数公开为bean属性。 很少需要直接使用ConcurrentTaskExecutor，但是如果ThreadPoolTaskExecutor不够灵活，无法满足您的需求，那么可以选择ConcurrentTaskExecutor。
 *  ThreadPoolTaskExecutor此实现是**最常用的实现**。 它公开了用于配置 java.util.concurrent.ThreadPoolExecutor的bean属性，并将其包装在TaskExecutor中。 如果需要适应其他类型的java.util.concurrent.Executor，建议您改用ConcurrentTaskExecutor。
 *  WorkManagerTaskExecutor此实现使用CommonJ WorkManager作为其支持服务提供者，并且是在Spring应用程序上下文中在WebLogic / WebSphere上设置基于CommonJ的线程池集成的中央便利类。
 *  DefaultManagedTaskExecutor此实现在兼容JSR-236的运行时环境（例如Java EE 7+应用程序服务器）中使用JNDI获得的ManagedExecutorService，为此替换了CommonJ WorkManager。

#### 使用TaskExecutor ####

Spring的TaskExecutor实现用作简单的JavaBean。 在下面的示例中，我们定义一个使用ThreadPoolTaskExecutor异步打印出一组消息的bean。

import org.springframework.beans.factory.InitializingBean;
    import org.springframework.core.task.TaskExecutor;
    
    /** * @author Created by niugang on 2020/4/7/20:46 */
    public class TaskExecutorExample implements InitializingBean { 
        @Override
        public void afterPropertiesSet() throws Exception { 
            printMessages();
        }
    
        private class MessagePrinterTask implements Runnable { 
    
            private String message;
    
            public MessagePrinterTask(String message) { 
                this.message = message;
            }
    
    
            @Override
            public void run() { 
                System.out.println(message);
            }
        }
    
        private TaskExecutor taskExecutor;
    
        public TaskExecutorExample(TaskExecutor taskExecutor) { 
            this.taskExecutor = taskExecutor;
        }
    
    
        public void printMessages() { 
            for (int i = 0; i < 25; i++) { 
                taskExecutor.execute(new MessagePrinterTask("Message" + i));
            }
        }
    }

如您所见，您可以将Runnable添加到队列，而TaskExecutor使用其内部规则来确定何时执行任务，而不是从池中检索线程并执行自己。

为了配置TaskExecutor将使用的规则，已经公开了简单的bean属性。

Java config

import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    /** * @author Created by niugang on 2020/4/8/14:41 */
    @Configuration
    public class ThreadPoolConfig { 
    
        @Bean
        public ThreadPoolTaskExecutor threadPoolTaskExecutor(){ 
            ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
            threadPool.setMaxPoolSize(10);
            threadPool.setCorePoolSize(2);
            threadPool.setQueueCapacity(100);
            return  threadPool;
        }
    
    }

import org.springframework.beans.factory.InitializingBean;
    import org.springframework.beans.factory.annotation.Autowired;
    
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.stereotype.Component;
    
    
    /** * @author Created by niugang on 2020/4/7/20:46 */
    @Component
    public class TaskExecutorExample implements InitializingBean { 
    
        private final ThreadPoolTaskExecutor threadPoolTaskExecutor;
    
        @Autowired
        public TaskExecutorExample(ThreadPoolTaskExecutor threadPoolTaskExecutor) { 
            this.threadPoolTaskExecutor = threadPoolTaskExecutor;
        }
    
        @Override
        public void afterPropertiesSet() throws Exception { 
            printMessages();
        }
    
        private class MessagePrinterTask implements Runnable { 
    
            private String message;
    
            public MessagePrinterTask(String message) { 
                this.message = message;
            }
    
    
            @Override
            public void run() { 
                System.out.println(message);
            }
        }
    
    
        public void printMessages() { 
            for (int i = 0; i < 25; i++) { 
                threadPoolTaskExecutor.execute(new MessagePrinterTask("Message" + i));
            }
        }
    }

### Spring TaskScheduler抽象 ###

除了TaskExecutor抽象之外，Spring 3.0还引入了TaskScheduler，它具有多种用于计划任务在将来某个时间点运行的方法。

public interface TaskScheduler { 
    
        ScheduledFuture schedule(Runnable task, Trigger trigger);
    
        ScheduledFuture schedule(Runnable task, Date startTime);
    
        ScheduledFuture scheduleAtFixedRate(Runnable task, Date startTime, long period);
    
        ScheduledFuture scheduleAtFixedRate(Runnable task, long period);
    
        ScheduledFuture scheduleWithFixedDelay(Runnable task, Date startTime, long delay);
    
        ScheduledFuture scheduleWithFixedDelay(Runnable task, long delay);
    }

最简单的方法是一个名为“schedule”的方法，它仅接受Runnable和Date。 这将导致任务在指定时间后运行一次。 所有其他方法都可以调度任务以重复运行。 固定速率和固定延迟方法用于简单的定期执行，但是接受触发器的方法则更加灵活。

#### Trigger接口 ####

Trigger接口本质上是受JSR-236（从Spring 3.0开始尚未正式实现）的启发。 触发器的基本思想是可以根据过去的执行结果甚至任意条件来确定执行时间。 如果这些确定确实考虑了先前执行的结果，则该信息在TriggerContext中可用。 Trigger接口本身非常简单：

public interface Trigger { 
    
        Date nextExecutionTime(TriggerContext triggerContext);
    }

如您所见，TriggerContext是最重要的部分。 它封装了所有相关数据，并在将来必要时开放以进行扩展。 TriggerContext是一个接口（默认情况下使用SimpleTriggerContext实现）。 在这里，您可以查看哪些方法可用于Trigger实现。

public interface TriggerContext { 
    
        Date lastScheduledExecutionTime();
    
        Date lastActualExecutionTime();
    
        Date lastCompletionTime();
    }

#### Trigger 实现 ####

Spring提供了Trigger接口的两种实现。 最有趣的是CronTrigger。 它启用了基于cron表达式的任务调度。 例如，以下任务计划在每小时的15分钟后运行，但仅在工作日的9到17个“工作时间”内运行。

scheduler.schedule(task, new CronTrigger("0 15 9-17 * * MON-FRI"));

![在这里插入图片描述][20200409163153935.png_pic_center]

另一种现成的实现是PeriodicTrigger，它接受一个固定的时间段，一个可选的初始延迟值和一个布尔值，以指示该时间段应解释为固定速率还是固定延迟。 由于TaskScheduler接口已经定义了以固定速率或固定延迟计划任务的方法，因此应尽可能直接使用这些方法。 PeriodicTrigger实现的价值在于它可以在依赖于触发器抽象的组件中使用。 例如，允许周期性触发器，基于cron的触发器，甚至自定义触发器实现可互换使用可能很方便。 这样的组件可以利用依赖注入的优势，从而可以在外部配置此类触发器，因此可以轻松地对其进行修改或扩展。

#### TaskScheduler实现 ####

![在这里插入图片描述][20200409163220187.png_pic_center]

与Spring的TaskExecutor抽象一样，TaskScheduler安排的主要好处是应用程序的调度需求与部署环境分离。 当部署到不应由应用程序本身直接创建线程的应用程序服务器环境时，此抽象级别特别重要。 对于此类情况，Spring提供了一个委托给WebLogic / WebSphere上的CommonJ TimerManager TimerManagerTaskScheduler，以及一个委托给Java EE 7+环境中的JSR-236 ManagedScheduledExecutorService的更新的DefaultManagedTaskScheduler，通常都配置了JNDI查找。

每当不需要外部线程管理时，一个更简单的选择是在应用程序中进行本地ScheduledExecutorService设置，可以通过Spring的ConcurrentTaskScheduler对其进行调整。 为了方便起见，Spring还提供了一个ThreadPoolTaskScheduler，它在内部委托给ScheduledExecutorService，从而按照ThreadPoolTaskExecutor的方式提供了通用的bean样式配置。 这些变体也适用于宽松的应用程序服务器环境中的本地嵌入式线程池设置，尤其是在Tomcat和Jetty上

import org.springframework.beans.factory.InitializingBean;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
    
    import java.util.Date;
    
    
    /** * @author Created by niugang on 2020/4/7/20:46 */
    
    public class TaskScheduleExample implements InitializingBean { 
    
       private   ThreadPoolTaskScheduler  threadPoolTaskScheduler;
    
        public TaskScheduleExample(ThreadPoolTaskScheduler threadPoolTaskScheduler) { 
            this.threadPoolTaskScheduler=threadPoolTaskScheduler;
        }
    
        @Override
        public void afterPropertiesSet() throws Exception { 
            threadPoolTaskScheduler.scheduleAtFixedRate(new Runnable() { 
                @Override
                public void run() { 
                     //5秒执行一次
                    System.out.println("定时任务执行："+new Date());
                }
            },new Date(),5000);
        }
    }

### 基于注解的任务计划和任务执行 ###

Spring为任务调度和异步方法执行提供注解支持

#### 启用计划任务注解 ####

要启用对@Scheduled和@Async批注的支持，请将@EnableScheduling和@EnableAsync添加到您的@Configuration类之一：

@Configuration
    @EnableAsync
    @EnableScheduling
    public class AppConfig { 
    }

您可以自由选择适合您的应用程序的相关注释。 例如，如果只需要对@Scheduled的支持，则只需省略@EnableAsync。 为了获得更细粒度的控制，您还可以实现**SchedulingConfigurer**或**AsyncConfigurer**接口

如果您更喜欢XML配置，请使用<task：annotation-driven>元素。

<task:annotation-driven executor="myExecutor" scheduler="myScheduler"/>
    <task:executor id="myExecutor" pool-size="5"/>
    <task:scheduler id="myScheduler" pool-size="10"/>

请注意，使用上面的XML，将提供执行程序引用来处理与带有@Async批注的方法相对应的那些任务，并且提供调度程序引用来管理以@Scheduled注释的那些方法。

**处理@Async批注的默认建议模式是“ proxy”，它仅允许通过代理来拦截呼叫。 同一类中的本地调用无法以这种方式被拦截。 对于更高级的侦听模式，请考虑结合编译时或加载时编织切换到“ aspectj”模式。**

注意：上面同一个类中无法进行异步执行

#### @Scheduled注解 ####

可以将@Scheduled注释与触发器元数据一起添加到方法中。 例如，以下方法将每隔5秒以固定的延迟被调用一次，这意味着将从每个先前调用的完成时间开始计算该时间段。

@Scheduled(fixedDelay=5000)
    public void doSomething() { 
        // something that should execute periodically
    }

如果需要固定速率执行，只需更改注释中指定的属性名称。 在每次调用的连续开始时间之间测量的每5秒执行一次以下操作。

@Scheduled(fixedRate=5000)
    public void doSomething() { 
        // something that should execute periodically
    }

对于固定延迟和固定速率的任务，可以指定初始延迟，以指示在第一次执行该方法之前要等待的毫秒数。

@Scheduled(initialDelay=1000, fixedRate=5000)
    public void doSomething() { 
        // something that should execute periodically
    }

如果简单的周期性调度不足以表现出来，则可以提供cron表达式。 例如，以下仅在工作日执行。

@Scheduled(cron="*/5 * * * * MON-FRI")
    public void doSomething() { 
        // something that should execute on weekdays only
    }

请注意，要调度的方法必须具有空返回值，并且不能期望任何参数。 如果该方法需要与应用程序上下文中的其他对象进行交互，则通常将通过依赖项注入来提供这些对象。

从Spring Framework 4.3开始，任何范围的bean都支持@Scheduled方法。

确保不要在运行时初始化同一@Scheduled注释类的多个实例，除非您确实希望为每个此类实例计划回调。 与此相关，请确保不要在使用@Scheduled注释并通过容器注册为常规Spring Bean的bean类上使用@Configurable：否则，您将获得双重初始化，一次通过容器，一次通过@Configurable方面 ，每个@Scheduled方法都会被调用两次。

#### @Async 注解 ####

可以在方法上提供@Async注解，以便对该方法的调用将异步发生。 换句话说，调用者将在调用后立即返回，并且该方法的实际执行将在已提交给Spring TaskExecutor的任务中发生。 在最简单的情况下，可以将注解应用于没有返回值的方法。

@Async
    void doSomething() { 
        // this will be executed asynchronously
    }

与用@Scheduled注解的方法不同，这些方法可以使用参数，因为它们将在运行时由调用者以“常规”方式调用，而不是从容器管理的计划任务中调用。 例如，以下是@Async注解的合法应用程序。

@Async
    void doSomething(String s) { 
        // this will be executed asynchronously
    }

@Async方法不仅可以声明常规的java.util.concurrent.Future返回类型，还可以声明Spring的org.springframework.util.concurrent.ListenableFuture，或者从Spring 4.2起，声明JDK 8的java.util.concurrent.CompletableFuture：以实现更丰富的交互 与异步任务并通过进一步的处理步骤立即合成。

异步实例：

import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.AsyncConfigurer;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.Executor;
    
    /** * @author Created by niugang on 2020/4/8/16:48 */
    @Configuration
    /** * 处理@Async批注的默认建议模式是“ proxy”，它仅允许通过代理来拦截呼叫。 * 同一类中的本地调用无法以这种方式被拦截。 * 对于更高级的侦听模式，请考虑结合编译时或加载时编织切换到“ aspectj”模式 */
    @EnableAsync
    public class AppAsyncConfig implements AsyncConfigurer { 
    
        @Override
        public Executor getAsyncExecutor() { 
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(7);
            executor.setMaxPoolSize(42);
            executor.setQueueCapacity(11);
            executor.setThreadNamePrefix("MyExecutor-");
            executor.initialize();
            return executor;
        }
    }

**controller**

@GetMapping("async")
        public Object findByCodeAndAuthor() { 
            asyncService.asyncMethod();
            return "success";
    
        }
    
        @GetMapping("async2")
        public Object findByCodeAndAuthor2() { 
            asyncService.asyncMethod2();
            return "success";
    
        }

**service**

import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationContext;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.stereotype.Service;
    
    import java.util.Date;
    import java.util.concurrent.TimeUnit;
    
    /** * @author Created by niugang on 2020/4/8/17:07 */
    @Service
    @Slf4j
    public class AsyncService { 
    
        private final ApplicationContext applicationContext;
    
        @Autowired
        public AsyncService(ApplicationContext applicationContext) { 
            this.applicationContext = applicationContext;
        }
    
    
        @Async
        public void asyncMethod() { 
    
    
            log.info("开始执行异步方法：{},Thread Name:{}", new Date(), Thread.currentThread().getName());
            try { 
                TimeUnit.SECONDS.sleep(5);
                log.info("异步方法执行结束：{}", new Date());
            } catch (InterruptedException e) { 
                Thread.currentThread().interrupt();
            }
    
        }
    
    
        /** * 解决同类方法调用 */
        public void asyncMethod2() { 
            AsyncService asyncService = applicationContext.getBean(AsyncService.class);
            asyncService.asyncMethod3();
    
        }
    
        @Async
        void asyncMethod3() { 
    
            log.info("开始执行异步本类方法：{},Thread Name:{}", new Date(), Thread.currentThread().getName());
            try { 
                TimeUnit.SECONDS.sleep(5);
                log.info("异步方法执行本类结束：{}", new Date());
            } catch (InterruptedException e) { 
                Thread.currentThread().interrupt();
            }
    
        }
    }

#### @Async的异常管理 ####

当@Async方法具有Future类型的返回值时，很容易管理在方法执行期间引发的异常，因为在调用Future结果时将引发此异常。 但是，对于void返回类型，该异常不会被捕获，并且无法传输。 对于这些情况，可以提供AsyncUncaughtExceptionHandler来处理此类异常。

public class MyAsyncUncaughtExceptionHandler implements AsyncUncaughtExceptionHandler { 
    
        @Override
        public void handleUncaughtException(Throwable ex, Method method, Object... params) { 
            // handle exception
        }
    }

默认情况下，仅记录异常。 可以通过AsyncConfigurer或task：annotation驱动的XML元素定义自定义AsyncUncaughtExceptionHandler。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70]

[20200409163153935.png_pic_center]: /images/20230528/55082ad10de04fa1b4a9dbc8e3be22fe.png
[20200409163220187.png_pic_center]: /images/20230528/4e65ce216f8a403c93dd8a6a04763eca.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70]: /images/20230528/e0701248d1bf4810bb8cb565cb00120a.png