当前位置:   article > 正文

线程池(三)----ThreadPoolTaskExecutor的提交方法execute和submit_threadpooltaskexecutor.submit

threadpooltaskexecutor.submit

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

无返回值的任务使用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接口:

后端打印:

主线程调用结束
MyOrderRunnableTask已执行

请求submit方法:

后端打印:

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

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

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

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

    举例:

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:

请求execute:

因为在两个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";
    }
请求:

验证确实没有捕获到。 

三、多线程与事务回滚:

上述,如果在事务中调用了多线程,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;
    }
}

请求

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

说明主线程回滚成功。 

(2)、主线程或子线程异常,主线程、子线程全部回滚:同时回滚主线程和子线程,就需要把主线程和子线程放到同一个事务中。

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

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

闽ICP备14008679号