Springboot Async schedule

/**
 * @Async异步方法线程池配置,默认不使用线程池,使用SimpleAsyncTaskExecutor(一个线程执行器,每个任务都会新建线程去执行)
 * 这里实现了接口AsyncConfigurer,并覆写了其内的方法,这样@Async默认的运行机制发生变化(使用了线程池,设置了线程运行过程的异常处理函数)
 * 备注:
 *   这里只是展示写法,要达到这个目的,可以不实现这个接口,具体见下面的方法
 * @DESC 
 * @author guchuang
 *
 */
@Configuration
public class AsyncConfig implements AsyncConfigurer {

    private static ExecutorService threadPool = new ThreadPoolExecutor(5, 5,
            60L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(3), new MyThreadFactory("common1"));
    
    private static ExecutorService threadPoolWithRejectDeal = new ThreadPoolExecutor(5, 5,
            60L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(3), new MyThreadFactory("common2"), new RejectedPolicy());
    
    /**
     * 这个实例声明的TaskExecutor会成为@Async方法运行的默认线程执行器
     * @Bean 使这个实例完全被spring接管
     */
    @Bean
    @Override
    public TaskExecutor getAsyncExecutor() {
        return new ConcurrentTaskExecutor(Executors.newFixedThreadPool(5,new MyThreadFactory("async1111")));
    }
    /**
     * 定义@Async方法默认的异常处理机制(只对void型异步返回方法有效,Future返回值类型的异常会抛给调用者)
     */
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (e, method, objects) -> MyLog.error("Method:" + method + ", exception:"+e.getMessage());
    }
    /**
     * 如果不覆写AsyncConfigurer的话,这个方法暴露bean会被当做@Async的默认线程池。
     * 注意必须是这个方法名(也就是bean name, 或者显示指定bean name @Qualifier("taskExecutor")),返回类型可以是Executor或者TaskExecutor
     * 如果没有配置的Executor,则默认使用SimpleAsyncTaskExecutor
     * 备注: 这种方式声明的bean,方法名就是bean name
     * @return
     */
    @Bean
    public Executor taskExecutor() {
        return new ConcurrentTaskExecutor(Executors.newFixedThreadPool(5,new MyThreadFactory("async0")));
    }
    /**
     * 定义其它的TaskExecutor,声明@Async方法的时候可以指定TaskExecutor,达到切换底层的目的
     * @return
     */
    @Bean
    public TaskExecutor async1() {
        return new ConcurrentTaskExecutor(Executors.newFixedThreadPool(2,new MyThreadFactory("async1")));
    }
    
    /**
     * 没有设置拒绝策略
     * @return
     */
    @Bean
    @Qualifier("async2")
    public TaskExecutor myAsyncExecutor2() {
        return new ConcurrentTaskExecutor(threadPool);
    }
    
    @Bean
    @Qualifier("async3")
    public TaskExecutor myAsyncExecutor3() {
        return new ConcurrentTaskExecutor(threadPoolWithRejectDeal);
    }
    
}
@Configuration
public class JndiConfig {
    @Bean
    @Qualifier("jndiTaskScheduler")
    TaskScheduler taskScheduler() {
        //正常情况下,jndi是寻找其它进程提供的服务,这里在本进程中注册只是为了测试效果
        registerJndi();
        //这个类会使用jndi从服务器上寻找服务,可能由其它jvm进程提供线程池。使用的名字为注册名:java:comp/DefaultManagedScheduledExecutorService
        return new DefaultManagedTaskScheduler();
    }

    /**
     * 使用jndi将线程池注入,下面的名字作为寻址的标识:java:comp/DefaultManagedScheduledExecutorService
     */

    public void registerJndi() {
        SimpleNamingContextBuilder b = new SimpleNamingContextBuilder();
        b.bind("java:comp/DefaultManagedScheduledExecutorService",
                Executors.newScheduledThreadPool(5));
        try {
            b.activate();
        } catch (IllegalStateException e) {
            e.printStackTrace();
        } catch (NamingException e) {
            e.printStackTrace();
        }
    }
}
@Configuration
public class ScheduleConfig implements SchedulingConfigurer{
    
    /**
     * 向spring容器注入TaskScheduler线程池,用于执行@Scheduled注解标注的方法.
     * 类型为TaskScheduler.class, name为taskExecutor1的bean(使用类型注入spring,不是bean name)
     * 如果没有注入TaskScheduler或者ScheduledExecutorService,则默认使用单线程的线程池作为底层支撑
     * @return TaskScheduler 实例
     */
    @Bean
    public TaskScheduler taskExecutor() {
        return new ConcurrentTaskScheduler(Executors.newScheduledThreadPool(3, new MyThreadFactory("scheduled")));
    }
    
    @Bean
    @Qualifier("test-123")
    public TaskScheduler taskExecutor2() {
        return new ConcurrentTaskScheduler(Executors.newScheduledThreadPool(3, new MyThreadFactory("scheduled2")));
    }

    /**
     * 可以用于执行定时任务,设置taskScheduler等
     */
    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        //taskRegistrar.setScheduler(taskExecutor1());      用于显示的设置线程池执行器
        taskRegistrar.addFixedDelayTask(() -> System.out.println("SchedulingConfigurer test"), 5000);
    }
    
}
/**
 *异步方法示例,关键点有三步:
 *  1.启动类增加注解 @EnableAsync
 *  2.当前类声明为服务 @Service
 *  3.方法上面添加注解 @Async
 *限制:
 *   默认类内的方法调用不会被aop拦截,也就是说同一个类内的方法调用,@Async不生效
 *解决办法:
 *  如果要使同一个类中的方法之间调用也被拦截,需要使用spring容器中的实例对象,而不是使用默认的this,因为通过bean实例的调用才会被spring的aop拦截
 *  本例使用方法: AsyncMethod asyncMethod = context.getBean(AsyncMethod.class);    然后使用这个引用调用本地的方法即可达到被拦截的目的
 *备注:
 *  这种方法只能拦截protected,default,public方法,private方法无法拦截。这个是spring aop的一个机制。
 *  
 * 默认情况下异步方法的调用使用的是SimpleAsyncTaskExecutor来执行异步方法调用,实际是每个方法都会起一个新的线程。
 * 大致运行过程:(以asyncMethod.bar1();为例)
 *  1.调用bar1()方法被aop拦截
 *  2.使用cglib获取要执行的方法和入参、当前实例(后续用于反射调用方法)。这些是运行一个方法的必要条件,可以封装成独立的方法来运行
 *  3.启动新的线程,调用上面封装的实际要调用的方法
 *  4.返回方法调用的结果
 *  前提是启动的时候被spring提前处理,将方法进行封装,加载流程:
 *    AsyncAnnotationBeanPostProcessor -> 
 * 如果要修改@Async异步方法底层调用:
 *  可以实现AsyncConfigurer接口,或者提供TaskExecutor实例(然后在@Async中指定这个实例),详见本例代码
 * 
 * 异步方法返回类型只能有两种:void和java.util.concurrent.Future
 *  当返回类型为void的时候,方法调用过程产生的异常不会抛到调用者层面,可以通过注册AsyncUncaughtExceptionHandler来捕获此类异常
 *  当返回类型为Future的时候,方法调用过程差生的异常会抛到调用者层面
 * 
 * @DESC 
 * @author guchuang
 *
 */
@Service
public class AsyncMethod {
    //@Autowired
    AsyncMethod asyncMethod;
    
    @Autowired
    WebApplicationContext context;
    
    @Async
    public void bar1() {
        MyLog.info("private bar");
    }
    @Async
    public void bar2() {
        MyLog.info("public bar");
    }
    @Async
    protected void bar3() {
        MyLog.info("protected bar");
    }
    @Async
    void bar4() {
        MyLog.info("default bar");
    }

    @Async
    public void foo1() {
        MyLog.info("foo1");
        this.bar1();
        this.bar2();
        asyncMethod = context.getBean(AsyncMethod.class);
        asyncMethod.bar1();     //同步
        asyncMethod.bar2();     //异步
        asyncMethod.bar3();     //异步
        asyncMethod.bar4();     //异步
    }
    
    /**
     * 指定这个异步方法使用的底层执行器TaskExecutor
     * @param index
     */
    @Async("async1")
    public void foo2(int index) {
        MyLog.info("foo2 with index:" + index);
       }
    
    @Async
    public void foo3(int index, String threadName) {
        Thread.currentThread().setName(threadName);
        MyLog.info("foo3 with index:" + index);
    }
    
    @Async
    public void fooE() {
        throw new RuntimeException("无返回值异步方法抛出异常");
    }
    @Async
    public Future<String> futureE() {
        throw new RuntimeException("有返回值异步方法抛出异常");
    }
    
    /**
     * 带返回值的异步调用
     * @return
     */
    @Async
    public Future<String> futureTask1() {
        MyLog.info("start run future task1");
        MyLog.sleep(1000);
        return new AsyncResult<String>("future task1");
    }
    @Async
    public CompletableFuture<String> futureTask2 () {
        MyLog.info("Running task  thread: " + Thread.currentThread().getName());

        CompletableFuture<String> future = new CompletableFuture<String>() {
            @Override
            public String get () throws InterruptedException, ExecutionException {
                return " task result";
            }
        };
        return future;
    }
    /**
     * 指定使用的TaskExecutor,这个bean在config中已经配置
     * @param index
     * @param time
     */
    @Async("async2")
    public void asyncSleep(int index, int time) {
        try {
            Thread.sleep(time * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        MyLog.info("task:" + index + " end");
    }
}
/**
 * 定时任务使用示例
 *  1.启动类增加注解 @EnableScheduling
 *  2.相应类声明为服务 @Service
 *  3.方法上面增加 @Scheduled, 指定不同的参数以不同的方式运行定时任务
 * 备注:
 *  @Scheduled中参数解释:
 *      fixedRate:表示以固定的时间间隔执行(上次开始执行和本次开始执行之间的时间间隔),不关注被执行方法实际的执行时间
 *      fixedDelay:表示以固定的延迟执行(上次执行结束和本次开始执行的时间间隔),受被执行方法执行时间的影响
 *      cron="2 * * * * *": cron表达式配置执行方法。总共6位,分别代表 秒 分 时 日 月 周
 *  spring底层执行器:(默认使用的是单线程线程池)
 *      this.localExecutor = Executors.newSingleThreadScheduledExecutor();
 *      this.taskScheduler = new ConcurrentTaskScheduler(this.localExecutor);
 *  加载流程:
 *      ScheduledAnnotationBeanPostProcessor -> ScheduledTaskRegistrar -> afterPropertiesSet()
 *  @Scheduled 的方法必须是空返回值和无参方法,直接调用这个方法是同步的
 * @DESC 
 * @author guchuang
 *
 */
@Service
public class ScheduleMethod {
    
    public ScheduleMethod() {
        MyLog.info("-----------------------ScheduleMethod init--------------------------");
    }
    
    @Scheduled(fixedRate=2000)
    public void foo3() {
        MyLog.info("每2秒执行一次,不管上次执行完成时间,fixedRate=2000");
        MyLog.sleep(1000);
    }
    @Scheduled(cron="*/2 * * * * *")
    public void foo5() {
        MyLog.info("cron test--每隔两秒执行一次");
    }
}
原文地址:https://www.cnblogs.com/gc65/p/12848544.html