devbox
爱喝兽奶帝天荒
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

线程池（三）ThreadPoolTaskExecutor（2）两种提交方法_threadpooltaskexecutor.execute

作者：爱喝兽奶帝天荒 | 2024-08-20 23:03:34

赞

踩

threadpooltaskexecutor.execute

ThreadPoolTaskExecutor有两种提交方法execute和submit：

无返回值的任务使用public void execute(Runnable command) 方法提交；

有返回值的任务使用public <T> Future<T> submit(Callable) 方法提交。

下面具体来看下两者的应用以及区别。

一、与主线程执行顺序的区别：

1、（1）public void execute(Runnable command) 方法提交,子线程可能在主线程结束之后结束；

举例：

  1.  @RequestMapping("/execute")
  2.     public String execute(){
  3.         System.out.println("进入方法");
  4.         threadPoolTaskExecutor.execute(new Runnable() {
  5.             @Override
  6.             public void run() {
  7.                 try {
  8.                     Thread.sleep(20000);
  9.                     System.out.println("sleep后");
  10.                 } catch (InterruptedException e) {
  11.                     e.printStackTrace();
  12.                 }
  13.             }
  14.         });
  15.         System.out.println("执行提交后");
  16.         return "aa";
  17.     }

请求后打印：

  1. 进入方法
  2. 执行提交后
  3. sleep后

可见，由于子线程比较耗时，主线程结束后子线程还没有执行完。 

（2）public <T> Future<T> submit(Callable) 方法提交，因为提交任务后有个取数据的过程，在从Future取数据的过程中，Callable自带的阻塞机制，这个机制保证主线程一定在子线程结束之后结束。反之如果没有取数据，子线程可能会在主线程结束之后才结束。

举例说明：

  1. package exceldemo.task;
  2.  
  3. import exceldemo.dto.User;
  4. import exceldemo.service.UserService;
  5.  
  6. import java.util.List;
  7. import java.util.concurrent.Callable;
  8.  
  9. public class UserTaskTest implements Callable<List<User>> {
  10.  
  11.     private List<Integer> ids;
  12.     private UserService userService;
  13.  
  14.     public UserTaskTest(List<Integer> childIds, UserService userService) {
  15.         System.out.println("构造");
  16.         this.ids = childIds;
  17.         this.userService = userService;
  18.     }
  19.  
  20.     @Override
  21.     public List<User> call() throws Exception {
  22.         Thread.sleep(4000);
  23.         System.out.println("执行");
  24.         return userService.getByIds(ids);
  25.     }
  26. }

例a：submit提交任务之后没有取数据：

  1. package exceldemo.rest;
  2.  
  3. import exceldemo.dto.User;
  4. import exceldemo.service.UserService;
  5. import exceldemo.task.UserTaskTest;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  8. import org.springframework.web.bind.annotation.RequestMapping;
  9. import org.springframework.web.bind.annotation.RestController;
  10.  
  11. import java.util.ArrayList;
  12. import java.util.List;
  13. import java.util.concurrent.Future;
  14.  
  15. @RestController
  16. @RequestMapping("/order")
  17. public class OrderTest {
  18.  
  19.     @Autowired
  20.     private ThreadPoolTaskExecutor threadPoolTaskExecutor;
  21.  
  22.     @Autowired
  23.     private UserService userService;
  24.  
  25.  
  26.     @RequestMapping("/submit")
  27.     public List<User> submit(){
  28.         List<Integer> ids = new ArrayList<>();
  29.         for(int i = 0;i<=500;i++){
  30.             ids.add(i);
  31.         }
  32.  
  33.         //异步获取所有用户
  34.  
  35.         List<User> users = new ArrayList<>();
  36.         List<Future> futures = new ArrayList<>();
  37.  
  38.         for (int i = 0; i < ids.size(); i += 100) {
  39.             int startIndex = i;
  40.             int endIndex = startIndex + 100 > ids.size() ? ids.size() : startIndex + 100;
  41.             UserTaskTest task = new UserTaskTest(ids.subList(startIndex, endIndex),userService);
  42.             Future<List<User>> future = threadPoolTaskExecutor.submit(task);
  43.             System.out.println("加入futurn");
  44.             futures.add(future);
  45.         }
  46.        
  47.         System.out.println("返回结果"+users.size());
  48.         return users;
  49.     }
  50. }

请求后后端打印：

  1. 构造
  2. 加入futurn
  3. 构造
  4. 加入futurn
  5. 构造
  6. 加入futurn
  7. 构造
  8. 加入futurn
  9. 构造
  10. 加入futurn
  11. 构造
  12. 加入futurn
  13. 返回结果0
  14. 执行
  15. 执行
  16. 执行
  17. 执行
  18. 执行
  19. 执行

 可以看到子线程比较耗时，主线程结束之后，子线程还没有执行完；

例b：submit提交任务之后取数据：

  1. @RequestMapping("/submit")
  2.     public List<User> submit(){
  3.         List<Integer> ids = new ArrayList<>();
  4.         for(int i = 0;i<=500;i++){
  5.             ids.add(i);
  6.         }
  7.  
  8.         //异步获取所有用户
  9.  
  10.         List<User> users = new ArrayList<>();
  11.         List<Future> futures = new ArrayList<>();
  12.  
  13.         for (int i = 0; i < ids.size(); i += 100) {
  14.             int startIndex = i;
  15.             int endIndex = startIndex + 100 > ids.size() ? ids.size() : startIndex + 100;
  16.             UserTaskTest task = new UserTaskTest(ids.subList(startIndex, endIndex),userService);
  17.             Future<List<User>> future = threadPoolTaskExecutor.submit(task);
  18.             System.out.println("加入futurn");
  19.             futures.add(future);
  20.         }
  21.         //取数据
  22.         try{
  23.             System.out.println("获取数据");
  24.             for(Future future : futures){
  25.                 System.out.println("获取数据内部");
  26.                 users.addAll((List<User>) future.get());
  27.             }
  28.         }catch (Exception e){
  29.  
  30.         }
  31.         System.out.println("返回结果"+users.size());
  32.         return users;
  33.     }

请求后后端打印：

  1. 构造
  2. 加入futurn
  3. 构造
  4. 加入futurn
  5. 构造
  6. 加入futurn
  7. 构造
  8. 加入futurn
  9. 构造
  10. 加入futurn
  11. 构造
  12. 加入futurn
  13. 获取数据
  14. 获取数据内部
  15. 执行
  16. 执行
  17. 执行
  18. 执行
  19. 执行
  20. 获取数据内部
  21. 获取数据内部
  22. 获取数据内部
  23. 获取数据内部
  24. 获取数据内部
  25. 执行
  26. 返回结果501

可以看到，即使子线程比主线程耗时，主线程也等子线程结束后才结束。

这两个例子证明了使用submit提交任务，提交后只要有从Future取数据的操作，就可以保证主线程在子线程结束后才结束。

************************************************************分割线****************************************************************************

2、下面再举个完整的例子，在子线程同样耗时以及主线程执行步骤一样的情况下比较execute和submit这两种方法：

线程池:

  1. package exceldemo.config;
  2.  
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  6.  
  7. @Configuration
  8. public class ThreadPoolTaskExecutorConfig {
  9.     private static int CORE_POOL_SIZE = 5;
  10.     private static int MAX_POOL_SIZE = 1000;
  11.     @Bean(name="threadPoolTaskExecutor")
  12.     public ThreadPoolTaskExecutor serviceJobTaskExecutor(){
  13.         ThreadPoolTaskExecutor poolTaskExecutor = new ThreadPoolTaskExecutor();
  14.         //线程池维护线程的最少数量
  15.         poolTaskExecutor.setCorePoolSize(CORE_POOL_SIZE);
  16.         //线程池维护线程的最大数量
  17.         poolTaskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
  18.         //线程池所使用的缓冲队列
  19.         poolTaskExecutor.setQueueCapacity(200);
  20.         //线程池维护线程所允许的空闲时间
  21.         poolTaskExecutor.setKeepAliveSeconds(30000);
  22.         poolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
  23.         System.out.println(poolTaskExecutor);
  24.         return poolTaskExecutor;
  25.     }
  26. }

controller接口： 

  1. package exceldemo.rest;
  2.  
  3. import exceldemo.task.MyOrderCallableTask;
  4. import exceldemo.task.MyOrderRunnableTask;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  7. import org.springframework.web.bind.annotation.RequestMapping;
  8. import org.springframework.web.bind.annotation.RestController;
  9.  
  10. import java.util.concurrent.ExecutionException;
  11. import java.util.concurrent.Future;
  12.  
  13. @RestController
  14. @RequestMapping("/myOrderDemo")
  15. public class MyOrderDemo {
  16.  
  17.     @Autowired
  18.     private ThreadPoolTaskExecutor threadPoolTaskExecutor;
  19.  
  20.     @RequestMapping("/execute")
  21.     public void execute(){
  22.         String str = "execute方法";
  23.         threadPoolTaskExecutor.execute(new MyOrderRunnableTask(str));
  24.         System.out.println("主线程调用结束");
  25.  
  26.     }
  27.  
  28.     @RequestMapping("/submit")
  29.     public String submit(){
  30.         String str = "submit方法";
  31.         Future<String> future = threadPoolTaskExecutor.submit(new MyOrderCallableTask(str));
  32.         try {
  33.             str = future.get();
  34.         } catch (InterruptedException e) {
  35.             e.printStackTrace();
  36.         } catch (ExecutionException e) {
  37.             e.printStackTrace();
  38.         }
  39.         System.out.println("主线程调用结束");
  40.         return str;
  41.     }
  42. }

Callable实现类： 

  1. package exceldemo.task;
  2.  
  3. import java.util.concurrent.Callable;
  4.  
  5. public class MyOrderCallableTask implements Callable<String> {
  6.  
  7.     private String name;
  8.  
  9.     public MyOrderCallableTask(String name) {
  10.         this.name = name;
  11.     }
  12.  
  13.     @Override
  14.     public String call() throws Exception {
  15.         try {
  16.             Thread.sleep(2000);
  17.         } catch (InterruptedException e) {
  18.             e.printStackTrace();
  19.         }
  20.         name = "MyOrderCallableTask";
  21.         System.out.println("MyOrderCallableTask已执行");
  22.         return name;
  23.     }
  24. }

Runnable实现类： 

  1. package exceldemo.task;
  2.  
  3. public class MyOrderRunnableTask implements Runnable{
  4.  
  5.     private String name;
  6.  
  7.     public MyOrderRunnableTask(String name){
  8.         this.name = name;
  9.     }
  10.  
  11.     @Override
  12.     public void run() {
  13.         try {
  14.             Thread.sleep(2000);
  15.         } catch (InterruptedException e) {
  16.             e.printStackTrace();
  17.         }
  18.         name = "MyOrderRunnableTask";
  19.         System.out.println("MyOrderRunnableTask已执行");
  20.     }
  21. }

启动类：

  1. package exceldemo;
  2. import org.springframework.boot.SpringApplication;
  3. import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Primary;
  7. import org.springframework.core.task.TaskExecutor;
  8. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  9.  
  10.  
  11. @SpringBootApplication
  12. public class Application {
  13. 	public static void main(String[] args) {
  14. 		SpringApplication.run(Application.class, args);
  15.  
  16. 	}
  17. }
  18.

 启动项目后，

请求execute接口:

后端打印：

  1. 主线程调用结束
  2. MyOrderRunnableTask已执行

请求submit方法：

后端打印：

  1. MyOrderCallableTask已执行
  2. 主线程调用结束

验证结束。这也和他们的功能是保持一致的。不需要返回结果，主线程就不需要等待子线程执行；需要返回结果，主线程肯定需要等所有的子线程结束后汇总结果。所以在调用的时候也需要注意：

（1）如果主线程调用了ThreadPoolTaskExecutor的execute提交任务，且传递了参数给子线程，并且子线程在修改这个参数，调用后主线程就不应该再使用这个参数，因为这个参数的值已经无法确定了；

（2）如果主线程调用了ThreadPoolTaskExecutor的submit提交任务，记得要在调用的逻辑后面，从Future里面把返回值取出来（调用Future的get方法），否则就和execute的效果一样了。

二、处理异常的区别：Callable执行call时遇到异常会抛出，而Runnable执行run时遇到异常并不会抛出。

    举例：

  1. package com.demo.rest;
  2.  
  3. import com.demo.service.UserService;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.web.bind.annotation.RequestMapping;
  6. import org.springframework.web.bind.annotation.RestController;
  7.  
  8. @RestController
  9. @RequestMapping("/user")
  10. public class UserController {
  11.  
  12.     @Autowired
  13.     private UserService userService;
  14.  
  15.     @RequestMapping("/submit")
  16.     public String submit(String param){
  17.  
  18.         param = userService.submit(param);
  19.         return param;
  20.     }
  21.  
  22.     @RequestMapping("/execute")
  23.     public String execute(String param){
  24.         String res = userService.execute(param);
  25.         return res;
  26.     }
  27. }
  1. package com.demo.service.impl;
  2.  
  3. import com.demo.dto.UserDTO;
  4. import com.demo.mapper.UserMapper;
  5. import com.demo.service.UserService;
  6. import com.demo.task.UserCallableTask;
  7. import com.demo.task.UserRunnableTask;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  10. import org.springframework.stereotype.Service;
  11. import org.springframework.transaction.annotation.Transactional;
  12.  
  13. import java.util.concurrent.ExecutionException;
  14. import java.util.concurrent.Future;
  15.  
  16. @Service("userService")
  17. public class UserServiceImpl implements UserService {
  18.  
  19.     @Autowired
  20.     private ThreadPoolTaskExecutor threadPoolTaskExecutor;
  21.  
  22.     @Autowired
  23.     private UserMapper userMapper;
  24.  
  25.     @Override
  26.     public String submit(String param) {
  27.  
  28.  
  29.         Future<String> future = threadPoolTaskExecutor.submit(new UserCallableTask(param));
  30.         try {
  31.             param =  future.get();
  32.         } catch (InterruptedException e) {
  33.             e.printStackTrace();
  34.             return "error";
  35.         } catch (ExecutionException e) {
  36.             e.printStackTrace();
  37.             return "error";
  38.         }
  39.         UserDTO user = new UserDTO();
  40.         user.setName(param);
  41.         userMapper.insert(user);
  42.         return param;
  43.     }
  44.  
  45.     @Override
  46.     public String execute(String param) {
  47.         threadPoolTaskExecutor.execute(new UserRunnableTask(param,userMapper));
  48.         return "success";
  49.     }
  50. }

task:

  1. package com.demo.task;
  2.  
  3. import com.demo.dto.UserDTO;
  4. import com.demo.mapper.UserMapper;
  5.  
  6. import java.util.concurrent.Callable;
  7.  
  8. public class UserCallableTask implements Callable<String> {
  9.  
  10.     private String param;
  11.  
  12.     public UserCallableTask (String param){
  13.         this.param = param;
  14.     }
  15.     @Override
  16.     public String call() throws Exception {
  17.         param += "UserCallableTask";
  18.         int a = 1/0;
  19.         return param;
  20.     }
  21. }
  1. package com.demo.task;
  2.  
  3. import com.demo.dto.UserDTO;
  4. import com.demo.mapper.UserMapper;
  5.  
  6. public class UserRunnableTask implements Runnable {
  7.  
  8.     private String param;
  9.  
  10.     private UserMapper userMapper;
  11.  
  12.     public UserRunnableTask (String param,UserMapper userMapper){
  13.         this.param = param;
  14.         this.userMapper = userMapper;
  15.     }
  16.  
  17.     @Override
  18.     public void run() {
  19.         param += "UserRunnableTask";
  20.         UserDTO user = new UserDTO();
  21.         user.setName(param);
  22.         int a = 1/0;
  23.         userMapper.insert(user);
  24.     }
  25. }

请求submit:

请求execute:

因为在两个task里面都加了异常1/0，所以请求这两个方法都不会往数据库插入数据。call方法抛出异常，service层捕获到后return就不再插入数据了，run方法自己遇到异常就终止了也不再往下执行。区别就再于二者对于异常的处理，调用sumbit方法执行时可以捕获异常，这样就可以自定义处理如把异常抛出给调用处（controller层），而execute的run方法遇到异常就自己终止了，主线程无法感知其运行成功与否。

有的人可能会想在调用execute时加上try...catch....，这个肯定是不可以的，这个try...catch...捕获的只是

threadPoolTaskExecutor.execute(new UserRunnableTask(param,userMapper));这个任务提交有没有异常，而这个任务和主线程是异步的，它实际执行的run方法主线程是捕获不到的。可以验证一下：
  1. @Override
  2.     @Transactional
  3.     public String execute(String param) {
  4.         try{
  5.             threadPoolTaskExecutor.execute(new UserRunnableTask(param,userMapper));
  6.         }catch (Exception e){
  7.             System.out.println("有异常");
  8.             return "error";
  9.         }
  10.  
  11.         return "success";
  12.     }

请求：

验证确实没有捕获到。 

三、多线程与事务回滚：

上述，如果在事务中调用了多线程，submit遇到异常会抛出且必须被捕获，不会触发回滚，execute遇到异常主线程无法感知，也不会触发回滚。那如果需要在多线程调用时实现事务回滚该怎么做呢？这就需要加入其它的操作了：

1、submit方法与事务回滚：我们知道sumbit方法提交线程在获取返回结果时是需要捕获异常的，那么我们就可以在捕获到异常时手动回滚当前事务。

（1）主线程正常，子线程发生异常，只回滚主线程：这种情况比较简单，主线程捕获异常后直接TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();回滚主线程就可以了：

  1. package com.demo.service.impl;
  2.  
  3. import com.demo.dto.UserDTO;
  4. import com.demo.mapper.UserMapper;
  5. import com.demo.service.UserService;
  6. import com.demo.task.UserCallableTask;
  7. import com.demo.task.UserRunnableTask;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  10. import org.springframework.stereotype.Service;
  11. import org.springframework.transaction.annotation.Transactional;
  12. import org.springframework.transaction.interceptor.TransactionAspectSupport;
  13.  
  14. import java.util.concurrent.ExecutionException;
  15. import java.util.concurrent.Future;
  16.  
  17. @Service("userService")
  18. public class UserServiceImpl implements UserService {
  19.  
  20.     @Autowired
  21.     private ThreadPoolTaskExecutor threadPoolTaskExecutor;
  22.  
  23.     @Autowired
  24.     private UserMapper userMapper;
  25.  
  26.     @Override
  27.     @Transactional
  28.     public String submit(String param) {
  29.  
  30.  
  31.         Future<String> future = threadPoolTaskExecutor.submit(new UserCallableTask(param,userMapper));
  32.         UserDTO user = new UserDTO();
  33.         user.setName("我是主线程");
  34.         userMapper.insert(user);
  35.         try {
  36.             param =  future.get();
  37.         } catch (InterruptedException e) {
  38.             e.printStackTrace();
  39.             TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
  40.             return "error";
  41.         } catch (ExecutionException e) {
  42.             e.printStackTrace();
  43.             TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
  44.             return "error";
  45.         }
  46.  
  47.         return param;
  48.     }
  49.  
  50.     @Override
  51.     public String execute(String param) {
  52.        // threadPoolTaskExecutor.execute(new UserRunnableTask(param,userMapper));
  53.         return "success";
  54.     }
  55. }

任务： 

  1. package com.demo.task;
  2.  
  3. import com.demo.dto.UserDTO;
  4. import com.demo.mapper.UserMapper;
  5.  
  6. import java.util.concurrent.Callable;
  7.  
  8. public class UserCallableTask implements Callable<String> {
  9.  
  10.     private String param;
  11.  
  12.     private UserMapper userMapper;
  13.  
  14.     public UserCallableTask(String param, UserMapper userMapper){
  15.         this.param = param;
  16.         this.userMapper = userMapper;
  17.     }
  18.     @Override
  19.     public String call() throws Exception {
  20.         param += "UserCallableTask";
  21.         UserDTO user = new UserDTO();
  22.         user.setName("我是子线程");
  23.         userMapper.insert(user);
  24.         int a = 1/0;
  25.         return param;
  26.     }
  27. }

请求

数据库没有主线程的数据插入：

说明主线程回滚成功。 

（2）、主线程或子线程异常，主线程、子线程全部回滚：同时回滚主线程和子线程，就需要把主线程和子线程放到同一个事务中。

说明主线程、子线程全部回滚成功。

2、execute方法：

四、 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/爱喝兽奶帝天荒/article/detail/1008838
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号