正文

Spring中的使用@Async异步调用方法

︶ㄣ米米素材网 メ
米米素材网V管理员 / 阅读
0705

这篇文章主要介绍了Spring中的使用@Async异步调用方法，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教

使用@Async异步调用方法

Async简介

异步方法调用使用场景：处理日志、发送邮件、短信......

spring中提供了@Async来实现异步方法。

@Async修饰类，则该类所有方法都是异步的，@Async修饰方法，则该方法是异步的。

被修饰的方法在被调用时，会在一个新的线程中执行。

Spring中通过在方法上设置@Async注解，可使得方法被异步调用。也就是该方法会在调用时立即返回，而这个方法的实际执行交

给Spring的TaskExecutor去完成

1. 如果此时线程池中的数量小于corePoolSize，即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。

2. 如果此时线程池中的数量等于corePoolSize，但是缓冲队列 workQueue未满，那么任务被放入缓冲队列。

3. 如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量小于maxPoolSize，建新的线程来处理被添加的任务。

4. 如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量等于maxPoolSize，那么通过handler所指定的策略来处理此任务。也就是：处理任务的优先级为：核心线程corePoolSize、任务队列workQueue、最大线程 maximumPoolSize，如果三者都满了，使用handler处理被拒绝的任务。

5. 当线程池中的线程数量大于corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池可以动态的调整池中的线程数。

本次记录Async使用场景

需要调用其他服务，并且主线程需要继续完成当前线程任务

第一步：需要去做事的类

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@Component
@EnableScheduling
public class VideoStatusUpdateServiceImpl implements VideoStatusUpdateService {
    @Resource
    private VaCaseVideoExtMapper vaCaseVideoExtMapper; 
    //每隔五秒
    @Scheduled(cron = "*/5 * * * * ? ")
    @Override
    public void videoStatusUpdate() throws IOException {
        //得到一个集合
        List<VaCaseVideo> list = vaCaseVideoExtMapper.selectAllVideoes();
        //遍历集合去创建异步线程，去做一些其他事情
        for (VaCaseVideo vo : list) {
            dealTask(vo);
        }
    }
    @Async("asyncServiceExecutor")
    public void dealTask(VaCaseVideo vo) throws IOException {
       System.out.print("这里在做某件事情")
    }  
}

第二步：启动类上加上注解@EnableAsync，开启异步

?
1
2
3
4
5
6
7
8
@SpringBootApplication
@EnableAsync
@EnableCaching
public class StartApp {
 public static void main(String[] args) {
  SpringApplication.run(StartApp.class, args);
 }
}

第三步：配置Executor（此步骤可有可无，若不配值则会使用默认值），配置自定义Executor

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@Configuration
public class ExecutorConfig {
    private static final Logger logger = LoggerFactory.getLogger(ExecutorConfig.class);
    @Bean
    public Executor asyncServiceExecutor() {
        logger.info("start asyncServiceExecutor");
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(5);
        //配置最大线程数
        executor.setMaxPoolSize(60);
        executor.setKeepAliveSeconds(180);
        //配置队列大小
        executor.setQueueCapacity(60);
        //配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix("async-service-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //执行初始化
        executor.initialize();
        return executor;
    }
}

第四步：启动项目，会每隔五秒打印需要做的事情

异步请求与异步调用的区别

两者的使用场景不同，异步请求用来解决并发请求对服务器造成的压力，从而提高对请求的吞吐量；而异步调用是用来做一些非主线流程且不需要实时计算和响应的任务，比如同步日志到kafka中做日志分析等。

异步请求是会一直等待response相应的，需要返回结果给客户端的；而异步调用我们往往会马上返回给客户端响应，完成这次整个的请求，至于异步调用的任务后台自己慢慢跑就行，客户端不会关心。

异步请求的实现

方式一：Servlet方式实现异步请求

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@RequestMapping(value = "/email/servletReq", method = GET)
public void servletReq (HttpServletRequest request, HttpServletResponse response) {
    AsyncContext asyncContext = request.startAsync();
    //设置监听器:可设置其开始、完成、异常、超时等事件的回调处理
    asyncContext.addListener(new AsyncListener() {
        @Override
        public void onTimeout(AsyncEvent event) throws IOException {
            System.out.println("超时了...");
            //做一些超时后的相关操作...
        }
        @Override
        public void onStartAsync(AsyncEvent event) throws IOException {
            System.out.println("线程开始");
        }
        @Override
        public void onError(AsyncEvent event) throws IOException {
            System.out.println("发生错误："+event.getThrowable());
        }
        @Override
        public void onComplete(AsyncEvent event) throws IOException {
            System.out.println("执行完成");
            //这里可以做一些清理资源的操作...
        }
    });
    //设置超时时间
    asyncContext.setTimeout(20000);
    asyncContext.start(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(10000);
                System.out.println("内部线程：" + Thread.currentThread().getName());
                asyncContext.getResponse().setCharacterEncoding("utf-8");
                asyncContext.getResponse().setContentType("text/html;charset=UTF-8");
                asyncContext.getResponse().getWriter().println("这是异步的请求返回");
            } catch (Exception e) {
                System.out.println("异常："+e);
            }
            //异步请求完成通知
            //此时整个请求才完成
            asyncContext.complete();
        }
    });
    //此时之类 request的线程连接已经释放了
    System.out.println("主线程：" + Thread.currentThread().getName());
}

方式二：使用很简单，直接返回的参数包裹一层callable即可，可以继承WebMvcConfigurerAdapter类来设置默认线程池和超时处理

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
  @RequestMapping(value = "/email/callableReq", method = GET)
  @ResponseBody
  public Callable<String> callableReq () {
      System.out.println("外部线程：" + Thread.currentThread().getName());
      return new Callable<String>() {
          @Override
          public String call() throws Exception {
              Thread.sleep(10000);
              System.out.println("内部线程：" + Thread.currentThread().getName());
              return "callable!";
          }
      };
  }
 
  @Configuration
  public class RequestAsyncPoolConfig extends WebMvcConfigurerAdapter {
 
  @Resource
  private ThreadPoolTaskExecutor myThreadPoolTaskExecutor;
 
  @Override
  public void configureAsyncSupport(final AsyncSupportConfigurer configurer) {
      //处理 callable超时
      configurer.setDefaultTimeout(60*1000);
      configurer.setTaskExecutor(myThreadPoolTaskExecutor);
      configurer.registerCallableInterceptors(timeoutCallableProcessingInterceptor());
  }
 
  @Bean
  public TimeoutCallableProcessingInterceptor timeoutCallableProcessingInterceptor() {
      return new TimeoutCallableProcessingInterceptor();
  }
}

方式三：和方式二差不多，在Callable外包一层，给WebAsyncTask设置一个超时回调，即可实现超时处理

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@RequestMapping(value = "/email/webAsyncReq", method = GET)
 @ResponseBody
 public WebAsyncTask<String> webAsyncReq () {
     System.out.println("外部线程：" + Thread.currentThread().getName());
     Callable<String> result = () -> {
         System.out.println("内部线程开始：" + Thread.currentThread().getName());
         try {
             TimeUnit.SECONDS.sleep(4);
         } catch (Exception e) {
             // TODO: handle exception
         }
         logger.info("副线程返回");
         System.out.println("内部线程返回：" + Thread.currentThread().getName());
         return "success";
     };
     WebAsyncTask<String> wat = new WebAsyncTask<String>(3000L, result);
     wat.onTimeout(new Callable<String>() {
 
         @Override
         public String call() throws Exception {
             // TODO Auto-generated method stub
             return "超时";
         }
     });
     return wat;
 }

方式四：DeferredResult可以处理一些相对复杂一些的业务逻辑，最主要还是可以在另一个线程里面进行业务处理及返回，即可在两个完全不相干的线程间的通信。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@RequestMapping(value = "/email/deferredResultReq", method = GET)
    @ResponseBody
    public DeferredResult<String> deferredResultReq () {
        System.out.println("外部线程：" + Thread.currentThread().getName());
        //设置超时时间
        DeferredResult<String> result = new DeferredResult<String>(60*1000L);
        //处理超时事件 采用委托机制
        result.onTimeout(new Runnable() {
 
            @Override
            public void run() {
                System.out.println("DeferredResult超时");
                result.setResult("超时了!");
            }
        });
        result.onCompletion(new Runnable() {
 
            @Override
            public void run() {
                //完成后
                System.out.println("调用完成");
            }
        });
        myThreadPoolTaskExecutor.execute(new Runnable() {
 
            @Override
            public void run() {
                //处理业务逻辑
                System.out.println("内部线程：" + Thread.currentThread().getName());
                //返回结果
                result.setResult("DeferredResult!!");
            }
        });
       return result;
    }

SpringBoot中异步调用的使用

1、介绍

异步请求的处理。除了异步请求，一般上我们用的比较多的应该是异步调用。通常在开发过程中，会遇到一个方法是和实际业务无关的，没有紧密性的。比如记录日志信息等业务。这个时候正常就是启一个新线程去做一些业务处理，让主线程异步的执行其他业务。

2、使用方式（基于spring下）

需要在启动类加入@EnableAsync使异步调用@Async注解生效

在需要异步执行的方法上加入此注解即可@Async("threadPool"),threadPool为自定义线程池

代码略。。。就俩标签，自己试一把就可以了

3、注意事项

在默认情况下，未设置TaskExecutor时，默认是使用SimpleAsyncTaskExecutor这个线程池，但此线程不是真正意义上的线程池，因为线程不重用，每次调用都会创建一个新的线程。可通过控制台日志输出可以看出，每次输出线程名都是递增的。所以最好我们来自定义一个线程池。

调用的异步方法，不能为同一个类的方法（包括同一个类的内部类），简单来说，因为Spring在启动扫描时会为其创建一个代理类，而同类调用时，还是调用本身的代理类的，所以和平常调用是一样的。

其他的注解如@Cache等也是一样的道理，说白了，就是Spring的代理机制造成的。所以在开发中，最好把异步服务单独抽出一个类来管理。下面会重点讲述。

4、什么情况下会导致@Async异步方法会失效？

  • a.调用同一个类下注有@Async异步方法：在spring中像@Async和@Transactional、cache等注解本质使用的是动态代理，其实Spring容器在初始化的时候Spring容器会将含有AOP注解的类对象“替换”为代理对象（简单这么理解），那么注解失效的原因就很明显了，就是因为调用方法的是对象本身而不是代理对象，因为没有经过Spring容器，那么解决方法也会沿着这个思路来解决。
  • b.调用的是静态(static )方法
  • c.调用(private)私有化方法

5、解决4中问题1的方式（其它2,3两个问题自己注意下就可以了）

将要异步执行的方法单独抽取成一个类，原理就是当你把执行异步的方法单独抽取成一个类的时候，这个类肯定是被Spring管理的，其他Spring组件需要调用的时候肯定会注入进去，这时候实际上注入进去的就是代理类了。

其实我们的注入对象都是从Spring容器中给当前Spring组件进行成员变量的赋值，由于某些类使用了AOP注解，那么实际上在Spring容器中实际存在的是它的代理对象。那么我们就可以通过上下文获取自己的代理对象调用异步方法。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Controller
@RequestMapping("/app")
public class EmailController {
    //获取ApplicationContext对象方式有多种,这种最简单,其它的大家自行了解一下
    @Autowired
    private ApplicationContext applicationContext;
 
    @RequestMapping(value = "/email/asyncCall", method = GET)
    @ResponseBody
    public Map<String, Object> asyncCall () {
        Map<String, Object> resMap = new HashMap<String, Object>();
        try{
            //这样调用同类下的异步方法是不起作用的
            //this.testAsyncTask();
            //通过上下文获取自己的代理对象调用异步方法
            EmailController emailController = (EmailController)applicationContext.getBean(EmailController.class);
            emailController.testAsyncTask();
            resMap.put("code",200);
        }catch (Exception e) {
            resMap.put("code",400);
            logger.error("error!",e);
        }
        return resMap;
    }
 
    //注意一定是public,且是非static方法
    @Async
    public void testAsyncTask() throws InterruptedException {
        Thread.sleep(10000);
        System.out.println("异步任务执行完成！");
    }
}

6、开启cglib代理，手动获取Spring代理类,从而调用同类下的异步方法。

首先，在启动类上加上@EnableAspectJAutoProxy(exposeProxy = true)注解。

代码实现，如下：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@Service
@Transactional(value = "transactionManager", readOnly = false, propagation = Propagation.REQUIRED, rollbackFor = Throwable.class)
public class EmailService {
 
    @Autowired
    private ApplicationContext applicationContext;
 
    @Async
    public void testSyncTask() throws InterruptedException {
        Thread.sleep(10000);
        System.out.println("异步任务执行完成！");
    }
    public void asyncCallTwo() throws InterruptedException {
        //this.testSyncTask();
//        EmailService emailService = (EmailService)applicationContext.getBean(EmailService.class);
//        emailService.testSyncTask();
        boolean isAop = AopUtils.isAopProxy(EmailController.class);//是否是代理对象；
        boolean isCglib = AopUtils.isCglibProxy(EmailController.class);  //是否是CGLIB方式的代理对象；
        boolean isJdk = AopUtils.isJdkDynamicProxy(EmailController.class);  //是否是JDK动态代理方式的代理对象；
        //以下才是重点!!!
        EmailService emailService = (EmailService)applicationContext.getBean(EmailService.class);
        EmailService proxy = (EmailService) AopContext.currentProxy();
        System.out.println(emailService == proxy ? true : false);
        proxy.testSyncTask();
        System.out.println("end!!!");
    }
}

以上为个人经验，希望能给大家一个参考，也希望大家多多支持米米素材网。

原文链接：https://blog.csdn.net/qq_34178998/article/details/95939425