线程池（四）----ThreadPoolTaskExecutor的提交方法execute和submit

电玩女神 2023-06-11 10:27 20阅读 0赞

前面提到了线程池提交任务有两种方法：

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

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

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

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

1、

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

`举例：`

@RequestMapping("/execute")
        public String execute(){
            System.out.println("进入方法");
            threadPoolTaskExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(20000);
                        System.out.println("sleep后");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            System.out.println("执行提交后");
            return "aa";
        }

请求后打印：

进入方法
    执行提交后
    sleep后

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

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

举例说明：

package exceldemo.task;
    
    import exceldemo.dto.User;
    import exceldemo.service.UserService;
    
    import java.util.List;
    import java.util.concurrent.Callable;
    
    public class UserTaskTest implements Callable<List<User>> {
    
        private List<Integer> ids;
        private UserService userService;
    
        public UserTaskTest(List<Integer> childIds, UserService userService) {
            System.out.println("构造");
            this.ids = childIds;
            this.userService = userService;
        }
    
        @Override
        public List<User> call() throws Exception {
            Thread.sleep(4000);
            System.out.println("执行");
            return userService.getByIds(ids);
        }
    }

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

package exceldemo.rest;
    
    import exceldemo.dto.User;
    import exceldemo.service.UserService;
    import exceldemo.task.UserTaskTest;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.Future;
    
    @RestController
    @RequestMapping("/order")
    public class OrderTest {
    
        @Autowired
        private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    
        @Autowired
        private UserService userService;
    
     
        @RequestMapping("/submit")
        public List<User> submit(){
            List<Integer> ids = new ArrayList<>();
            for(int i = 0;i<=500;i++){
                ids.add(i);
            }
    
            //异步获取所有用户
    
            List<User> users = new ArrayList<>();
            List<Future> futures = new ArrayList<>();
    
            for (int i = 0; i < ids.size(); i += 100) {
                int startIndex = i;
                int endIndex = startIndex + 100 > ids.size() ? ids.size() : startIndex + 100;
                UserTaskTest task = new UserTaskTest(ids.subList(startIndex, endIndex),userService);
                Future<List<User>> future = threadPoolTaskExecutor.submit(task);
                System.out.println("加入futurn");
                futures.add(future);
            }
           
            System.out.println("返回结果"+users.size());
            return users;
        }
    }

请求后后端打印：

构造
    加入futurn
    构造
    加入futurn
    构造
    加入futurn
    构造
    加入futurn
    构造
    加入futurn
    构造
    加入futurn
    返回结果0
    执行
    执行
    执行
    执行
    执行
    执行

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

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

@RequestMapping("/submit")
        public List<User> submit(){
            List<Integer> ids = new ArrayList<>();
            for(int i = 0;i<=500;i++){
                ids.add(i);
            }
    
            //异步获取所有用户
    
            List<User> users = new ArrayList<>();
            List<Future> futures = new ArrayList<>();
    
            for (int i = 0; i < ids.size(); i += 100) {
                int startIndex = i;
                int endIndex = startIndex + 100 > ids.size() ? ids.size() : startIndex + 100;
                UserTaskTest task = new UserTaskTest(ids.subList(startIndex, endIndex),userService);
                Future<List<User>> future = threadPoolTaskExecutor.submit(task);
                System.out.println("加入futurn");
                futures.add(future);
            }
            //取数据
            try{
                System.out.println("获取数据");
                for(Future future : futures){
                    System.out.println("获取数据内部");
                    users.addAll((List<User>) future.get());
                }
            }catch (Exception e){
    
            }
            System.out.println("返回结果"+users.size());
            return users;
        }

请求后后端打印：

构造
    加入futurn
    构造
    加入futurn
    构造
    加入futurn
    构造
    加入futurn
    构造
    加入futurn
    构造
    加入futurn
    获取数据
    获取数据内部
    执行
    执行
    执行
    执行
    执行
    获取数据内部
    获取数据内部
    获取数据内部
    获取数据内部
    获取数据内部
    执行
    返回结果501

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

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

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

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

线程池:

package exceldemo.config;
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    @Configuration
    public class ThreadPoolTaskExecutorConfig {
        private static int CORE_POOL_SIZE = 5;
        private static int MAX_POOL_SIZE = 1000;
        @Bean(name="threadPoolTaskExecutor")
        public ThreadPoolTaskExecutor serviceJobTaskExecutor(){
            ThreadPoolTaskExecutor poolTaskExecutor = new ThreadPoolTaskExecutor();
            //线程池维护线程的最少数量
            poolTaskExecutor.setCorePoolSize(CORE_POOL_SIZE);
            //线程池维护线程的最大数量
            poolTaskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
            //线程池所使用的缓冲队列
            poolTaskExecutor.setQueueCapacity(200);
            //线程池维护线程所允许的空闲时间
            poolTaskExecutor.setKeepAliveSeconds(30000);
            poolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
            System.out.println(poolTaskExecutor);
            return poolTaskExecutor;
        }
    }

controller接口：

package exceldemo.rest;
    
    import exceldemo.task.MyOrderCallableTask;
    import exceldemo.task.MyOrderRunnableTask;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Future;
    
    @RestController
    @RequestMapping("/myOrderDemo")
    public class MyOrderDemo {
    
        @Autowired
        private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    
        @RequestMapping("/execute")
        public void execute(){
            String str = "execute方法";
            threadPoolTaskExecutor.execute(new MyOrderRunnableTask(str));
            System.out.println("主线程调用结束");
    
        }
    
        @RequestMapping("/submit")
        public String submit(){
            String str = "submit方法";
            Future<String> future = threadPoolTaskExecutor.submit(new MyOrderCallableTask(str));
            try {
                str = future.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
            System.out.println("主线程调用结束");
            return str;
        }
    }

Callable实现类：

package exceldemo.task;
    
    import java.util.concurrent.Callable;
    
    public class MyOrderCallableTask implements Callable<String> {
    
        private String name;
    
        public MyOrderCallableTask(String name) {
            this.name = name;
        }
    
        @Override
        public String call() throws Exception {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            name = "MyOrderCallableTask";
            System.out.println("MyOrderCallableTask已执行");
            return name;
        }
    }

Runnable实现类：

package exceldemo.task;
    
    public class MyOrderRunnableTask implements Runnable{
    
        private String name;
    
        public MyOrderRunnableTask(String name){
            this.name = name;
        }
    
        @Override
        public void run() {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            name = "MyOrderRunnableTask";
            System.out.println("MyOrderRunnableTask已执行");
        }
    }

启动类：

package exceldemo;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Primary;
    import org.springframework.core.task.TaskExecutor;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    
    @SpringBootApplication
    public class Application {
    	public static void main(String[] args) {
    		SpringApplication.run(Application.class, args);
     
    	}
    }

启动项目后，

请求execute接口:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70][]

后端打印：

主线程调用结束
    MyOrderRunnableTask已执行

请求submit方法：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 1][]

后端打印：

MyOrderCallableTask已执行
    主线程调用结束

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

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

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

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

举例：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 2][]

package com.demo.rest;
    
    import com.demo.service.UserService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    @RequestMapping("/user")
    public class UserController {
    
        @Autowired
        private UserService userService;
    
        @RequestMapping("/submit")
        public String submit(String param){
    
            param = userService.submit(param);
            return param;
        }
    
        @RequestMapping("/execute")
        public String execute(String param){
            String res = userService.execute(param);
            return res;
        }
    }

package com.demo.service.impl;
    
    import com.demo.dto.UserDTO;
    import com.demo.mapper.UserMapper;
    import com.demo.service.UserService;
    import com.demo.task.UserCallableTask;
    import com.demo.task.UserRunnableTask;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Future;
    
    @Service("userService")
    public class UserServiceImpl implements UserService {
    
        @Autowired
        private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    
        @Autowired
        private UserMapper userMapper;
    
        @Override
        public String submit(String param) {
    
    
            Future<String> future = threadPoolTaskExecutor.submit(new UserCallableTask(param));
            try {
                param =  future.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
                return "error";
            } catch (ExecutionException e) {
                e.printStackTrace();
                return "error";
            }
            UserDTO user = new UserDTO();
            user.setName(param);
            userMapper.insert(user);
            return param;
        }
    
        @Override
        public String execute(String param) {
            threadPoolTaskExecutor.execute(new UserRunnableTask(param,userMapper));
            return "success";
        }
    }

task:

package com.demo.task;
    
    import com.demo.dto.UserDTO;
    import com.demo.mapper.UserMapper;
    
    import java.util.concurrent.Callable;
    
    public class UserCallableTask implements Callable<String> {
    
        private String param;
    
        public UserCallableTask (String param){
            this.param = param;
        }
        @Override
        public String call() throws Exception {
            param += "UserCallableTask";
            int a = 1/0;
            return param;
        }
    }

package com.demo.task;
    
    import com.demo.dto.UserDTO;
    import com.demo.mapper.UserMapper;
    
    public class UserRunnableTask implements Runnable {
    
        private String param;
    
        private UserMapper userMapper;
    
        public UserRunnableTask (String param,UserMapper userMapper){
            this.param = param;
            this.userMapper = userMapper;
        }
    
        @Override
        public void run() {
            param += "UserRunnableTask";
            UserDTO user = new UserDTO();
            user.setName(param);
            int a = 1/0;
            userMapper.insert(user);
        }
    }

请求submit:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 3][]

请求execute:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 4][]

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

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

threadPoolTaskExecutor.execute(new UserRunnableTask(param,userMapper));这个任务提交有没有异常，而这个任务和主线程是异步的，它实际执行的run方法主线程是捕获不到的。可以验证一下：

@Override
        @Transactional
        public String execute(String param) {
            try{
                threadPoolTaskExecutor.execute(new UserRunnableTask(param,userMapper));
            }catch (Exception e){
                System.out.println("有异常");
                return "error";
            }
    
            return "success";
        }

请求：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 5][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 6][]

验证确实没有捕获到。

三、多线程与事务回滚：

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

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

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

package com.demo.service.impl;
    
    import com.demo.dto.UserDTO;
    import com.demo.mapper.UserMapper;
    import com.demo.service.UserService;
    import com.demo.task.UserCallableTask;
    import com.demo.task.UserRunnableTask;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;
    import org.springframework.transaction.interceptor.TransactionAspectSupport;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Future;
    
    @Service("userService")
    public class UserServiceImpl implements UserService {
    
        @Autowired
        private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    
        @Autowired
        private UserMapper userMapper;
    
        @Override
        @Transactional
        public String submit(String param) {
    
    
            Future<String> future = threadPoolTaskExecutor.submit(new UserCallableTask(param,userMapper));
            UserDTO user = new UserDTO();
            user.setName("我是主线程");
            userMapper.insert(user);
            try {
                param =  future.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
                TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                return "error";
            } catch (ExecutionException e) {
                e.printStackTrace();
                TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                return "error";
            }
    
            return param;
        }
    
        @Override
        public String execute(String param) {
           // threadPoolTaskExecutor.execute(new UserRunnableTask(param,userMapper));
            return "success";
        }
    }

任务：

package com.demo.task;
    
    import com.demo.dto.UserDTO;
    import com.demo.mapper.UserMapper;
    
    import java.util.concurrent.Callable;
    
    public class UserCallableTask implements Callable<String> {
    
        private String param;
    
        private UserMapper userMapper;
    
        public UserCallableTask(String param, UserMapper userMapper){
            this.param = param;
            this.userMapper = userMapper;
        }
        @Override
        public String call() throws Exception {
            param += "UserCallableTask";
            UserDTO user = new UserDTO();
            user.setName("我是子线程");
            userMapper.insert(user);
            int a = 1/0;
            return param;
        }
    }

请求

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 7][]

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

![20191101111726197.png][]

说明主线程回滚成功。

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

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

2、execute方法：

四、

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70]: /images/20230531/af9dfc792b5b46c3968108171160e63d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 1]: /images/20230531/1faeb7ecbb65450e99b3e83088dc8dad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 2]: /images/20230531/0be66ca430534077ab28d117cbeaff03.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 3]: /images/20230531/efdffde3232c4c14a6e64ac23fc79b4f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 4]: /images/20230531/b7a7034fd2cf4f38b6446f241cf4e397.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 5]: /images/20230531/71ad176dbafd44649c5d5e13caa84924.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 6]: /images/20230531/e4a8c01b531545cdb8f1e63754bfb8c2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70 7]: /images/20230531/e66941901e6d436cba15cd60cd0afe8e.png
[20191101111726197.png]: /images/20230531/5cf95ad73218494c80295a57bb2d7cf2.png