线程池（三）----ThreadPoolTaskExecutor

柔光的暖阳◎ 2023-05-31 12:26 17阅读 0赞

应用举例：假设有三个耗时任务，一个返回结果，一个不返回结果，一个不返回结果且批量执行

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70][]

1、ThreadPoolTaskExecutorConfig：

将ThreadPoolTaskExecutor做为一个bean，通过spring的注入，保证只会初始化一次。

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;
        }
    }

2、服务：

package exceldemo.service.impl;
    
    import exceldemo.dto.User;
    import exceldemo.service.UserService;
    import org.springframework.stereotype.Service;
    
    import java.util.ArrayList;
    import java.util.List;
    
    @Service("userService")
    public class UserServiceImpl implements UserService {
        @Override
        public List<User> getByIds(List<Integer> ids) {
    
            List<User> users = new ArrayList<>();
            for(Integer id : ids){
                User user = new User();
                user.setAge(id);
                user.setUserName("用户"+id);
                //耗时操作
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                users.add(user);
            }
            return users;
        }
    
        @Override
        public void addUserAction() {
    
            //耗时操作
            try {
                Thread.sleep(20);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("行为记录");
        }
    }

3、需要提交的task：

package exceldemo.task;
    
    import exceldemo.dto.User;
    import exceldemo.service.UserService;
    
    import java.util.List;
    import java.util.concurrent.Callable;
    
    public class UserTask  implements Callable<List<User>> {
    
        private List<Integer> userIds;
    
        private UserService userService;
    
        public UserTask(List<Integer> queryIds, UserService userService) {
            this.userService = userService;
            this.userIds = queryIds;
            //System.out.println("ids  "+queryIds.size());
        }
    
    
        @Override
        public List<User> call() throws Exception {
            //System.out.println("*******************");
            List<User> users = userService.getByIds(userIds);
            return users;
        }
    }

异步任务：

package exceldemo.task;
    
    import java.util.List;
    
    public class DemoTask implements Runnable {
    
        private List<Integer> ids ;
    
        public  DemoTask(List<Integer> ids){
            this.ids = ids;
        }
    
        @Override
        public void run() {
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //System.out.println("run "+ids.size());
        }
    }

4、controller：

package exceldemo.rest;
    
    import exceldemo.dto.User;
    import exceldemo.service.UserService;
    import exceldemo.task.UserTask;
    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.Date;
    import java.util.List;
    import java.util.concurrent.Future;
    import java.util.concurrent.ThreadPoolExecutor;
    
    @RestController
    @RequestMapping("/user1")
    public class UserController {
    
        @Autowired
        private UserService userService;
    
        @Autowired
        private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    
        @RequestMapping("/getAllSc")
        public List<User> getAllSc(){
           
            List<Integer> ids = new ArrayList<>();
            for(int i = 0;i<=500;i++){
                ids.add(i);
            }
            //1、异步通知
            threadPoolTaskExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    userService.addUserAction();
                }
            });
    
            //2、批量异步通知
            List<Integer> childIds = new ArrayList<>();
            for (int i = 0; i < ids.size(); i += 100) {
                int startIndex = i;
                int endIndex = startIndex + 100 > ids.size() ? ids.size() : startIndex + 100;
                DemoTask task = new DemoTask(ids.subList(startIndex, endIndex));
               threadPoolTaskExecutor.execute(task);
            }
    
            //3、异步获取所有用户
            long startTime = new Date().getTime();
            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;
                UserTask task = new UserTask(ids.subList(startIndex, endIndex),userService);
                Future<List<User>> future = threadPoolTaskExecutor.submit(task);
                futures.add(future);
            }
            //取数据
            try{
                for(Future future : futures){
                    users.addAll((List<User>) future.get());
                }
            }catch (Exception e){
    
            }
    
            long endTime = new Date().getTime();
            System.out.println("耗时"+(endTime-startTime));
            return users;
        }
    }

控制台打印：

行为记录
    耗时1046

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dfdF95X3k_size_16_color_FFFFFF_t_70]: /images/20230531/76159d7e566e4f1387d2f91068d0052f.png