千万级别的批量查询

客官°小女子只卖身不卖艺 2023-06-25 03:04 3阅读 0赞

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
                <exclusions>
                    <exclusion>
                        <groupId>org.junit.vintage</groupId>
                        <artifactId>junit-vintage-engine</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>com.google.guava</groupId>
                <artifactId>guava</artifactId>
                <version>22.0</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.10</version>
            </dependency>

package com.citydo.checkandbigdataquery.controller;
    
    
    import com.citydo.checkandbigdataquery.model.Task;
    import com.citydo.checkandbigdataquery.service.TaskService;
    import com.citydo.checkandbigdataquery.task.TaskQuery;
    import com.google.common.collect.Lists;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.http.ResponseEntity;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.util.ObjectUtils;
    import org.springframework.web.bind.annotation.CrossOrigin;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    import javax.annotation.Resource;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.*;
    
    
    @CrossOrigin
    @RestController
    public class TaskController { 
    
        @Resource(name = "defaultThreadPool")
        private ThreadPoolTaskExecutor executor;
    
        @Autowired
        private TaskService taskService;
    
        @GetMapping("/query")
        public ResponseEntity query(){ 
    
            // 模拟10000个id
            List<Long> ids = new ArrayList<>();
            for (long i = 0; i <10000000; i++) { 
                ids.add(i);
            }
    
            // 将ids按多少个进行拆分
            List<List<Long>> parts = Lists.partition(ids, 50);
    
            // 有多少份，就有多少次请求，有多少次请求就有多少个task
            List<Callable<List<Task>>> tasks = new ArrayList<>();
            for (List<Long> part : parts) { 
                TaskQuery taskQuery = new TaskQuery(taskService);
                taskQuery.setIds(part);
                tasks.add(taskQuery);
            }
            // 每一个任务（线程）执行的结果都是一个future对象，这个对象的数据就是List<Task>
            List<Future<List<Task>>> futures = new ArrayList<>();
            for (Callable<List<Task>> task : tasks) { 
                // 提交任务 == 注意这时候并没有触发线程执行，就绪状态
                futures.add(executor.submit(task));
            }
            // 最终要查询的任务集
            List<Task> allTasks = new ArrayList<>();
            // 遍历执行结果，准备正式执行（get）
            for (Future<List<Task>> future : futures) { 
                try { 
                    // 查询任务超过3s的直接弃掉（cancel取消任务）
                    List<Task> taskList =future.get(3000,TimeUnit.MILLISECONDS);
                    if(!ObjectUtils.isEmpty(taskList)){ 
                        allTasks.addAll(taskList);
                    }
                }catch (InterruptedException | ExecutionException | TimeoutException e){ 
                    future.cancel(true);
                }
            }
            return ResponseEntity.ok(allTasks);
        }
    
        @GetMapping("/queryNot")
        public ResponseEntity queryNot(){ 
            // 模拟10000个id
            List<Long> ids = new ArrayList<>();
            for (long i = 0; i <10000000; i++) { 
                ids.add(i);
            }
            return ResponseEntity.ok(taskService.queryByIds(ids));
        }
    
    }

package com.citydo.checkandbigdataquery.config;
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import java.util.UUID;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /** * 线程池配置 */
    @Configuration
    public class ThreadPoolExecutorConfig { 
    
        private int corePoolSize = 10;
    
        private int maxPoolSize = 30;
    
        private int queueCapacity = 8;
    
        private int keepAlive = 60;
    
        /** * 默认线程池线程池 * * @return Executor */
        @Bean
        public ThreadPoolTaskExecutor defaultThreadPool() { 
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            //核心线程数目
            executor.setCorePoolSize(corePoolSize);
            //指定最大线程数
            executor.setMaxPoolSize(maxPoolSize);
            //队列中最大的数目
            executor.setQueueCapacity(queueCapacity);
            //线程名称前缀
            executor.setThreadNamePrefix("defaultThreadPool_"+ UUID.randomUUID().toString());
            //rejection-policy：当pool已经达到max size的时候，如何处理新任务
            //CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
            //对拒绝task的处理策略
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            //线程空闲后的最大存活时间
            executor.setKeepAliveSeconds(keepAlive);
            //加载
            executor.initialize();
            return executor;
        }
    
    }

package com.citydo.checkandbigdataquery.service;
    
    import com.citydo.checkandbigdataquery.model.Task;
    import org.springframework.stereotype.Service;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.UUID;
    
    /** * <p>任务接口实现</p> */
    @Service
    public class TaskService { 
    
        public List<Task> queryByIds(List<Long> ids){ 
            List<Task> tasks = new ArrayList<>();
            for (Long id : ids) { 
                tasks.add(new Task(id, UUID.randomUUID().toString()));
            }
            return tasks;
        }
    }

package com.citydo.checkandbigdataquery.task;
    
    
    import com.citydo.checkandbigdataquery.model.Task;
    import com.citydo.checkandbigdataquery.service.TaskService;
    import lombok.Data;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.Callable;
    
    
    /** * <p>任务查询线程</p> */
    @Data
    public class TaskQuery implements Callable<List<Task>> { 
    
        private List<Long> ids ;
    
        private TaskService taskService;
    
        /** * 注入service * @param taskService */
        public TaskQuery(TaskService taskService) { 
            ids = new ArrayList<>();
            this.taskService = taskService;
        }
    
        /** * 回调查询 * @return */
        @Override
        public List<Task> call(){ 
            return taskService.queryByIds(this.ids);
        }
    
    }

package com.citydo.checkandbigdataquery.model;
    
    import lombok.Data;
    
    /** * @author nick */
    @Data
    public class Task { 
        private Long id;
        private String name;
    
        public Task(Long id, String name) { 
            this.id = id;
            this.name = name;
        }
    }

参考：[https://blog.csdn.net/Appleyk][https_blog.csdn.net_Appleyk]

[https_blog.csdn.net_Appleyk]: https://blog.csdn.net/Appleyk