# 需求: 公司cms系统需要开发针对企业信息的高级搜索功能, 需要将mysql的数据导入的es。 主表是2亿多, 同步es大约需要查询40个表  由于数据量大， 组成一条es数据就需要查询40个表， 这里使用线程池（ThreadPoolTaskExecutor） + 异步框架（CompletableFuture） 数据库能使用覆盖索引就使用, 必须要加索引, 手写sql, 不要用mybatis plus 这些框架提供的查询, 它那个查询返回的都是全字段, 这里要select 具体字段, 提高下性能(因为数据量太大, 要只返回个必须的字段, 而且尽量使用覆盖索引) # 配置线程池 ThreadPoolTaskExecutor ThreadPoolConfig .class ``` import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.concurrent.ThreadPoolExecutor; @Configuration @EnableAsync @Slf4j public class ThreadPoolConfig { @Value("${async.executor.thread.core_pool_size}") private int corePoolSize; @Value("${async.executor.thread.max_pool_size}") private int maxPoolSize; @Value("${async.executor.thread.queue_capacity}") private int queueCapacity; // @Value("${async.executor.thread.name.prefix}") // private String namePrefix; @Value("${async.executor.thread.keep_alive_seconds}") private int keepAliveSeconds; /** * IO密集型任务 = 一般为2*CPU核心数（常出现于线程中：数据库数据交互、文件上传下载、网络数据传输等等） * CPU密集型任务 = 一般为CPU核心数+1（常出现于线程中：复杂算法） * 混合型任务 = 视机器配置和复杂度自测而定 */ @Bean(name = "asyncTaskExecutor") public ThreadPoolTaskExecutor asyncTaskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); // ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor(); //1: 核心线程数目 executor.setCorePoolSize(corePoolSize); //2: 指定最大线程数,只有在缓冲队列满了之后才会申请超过核心线程数的线程 executor.setMaxPoolSize(maxPoolSize); //3: 队列中最大的数目 executor.setQueueCapacity(queueCapacity); //4: 线程名称前缀 // executor.setThreadNamePrefix(namePrefix); //5:当pool已经达到max size的时候，如何处理新任务 // CallerRunsPolicy: 会在execute 方法的调用线程中运行被拒绝的任务,如果执行程序已关闭，则会丢弃该任务 // AbortPolicy: 抛出java.util.concurrent.RejectedExecutionException异常 // DiscardOldestPolicy: 抛弃旧的任务 // DiscardPolicy: 抛弃当前的任务 // executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy() { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { System.out.println("丢弃"); } }); //6: 线程空闲后的最大存活时间(默认值 60),当超过了核心线程出之外的线程在空闲时间到达之后会被销毁 executor.setKeepAliveSeconds(keepAliveSeconds); //7:线程空闲时间,当线程空闲时间达到keepAliveSeconds(秒)时,线程会退出,直到线程数量等于corePoolSize,如果allowCoreThreadTimeout=true,则会直到线程数量等于0 executor.setAllowCoreThreadTimeOut(false); executor.initialize(); return executor; } } ``` yml ``` async: executor: thread: # 核心线程数目 core_pool_size: 5 # 最大线程数 max_pool_size: 10 # 队列中最大的数目 queue_capacity: 99999 # 秒 线程空闲后的最大存活时间 keep_alive_seconds: 600 ``` # 异步 CompletableFuture ``` 部分查询方法 // 导入线程池 @Autowired private AsyncTaskExecutor asyncTaskExecutor; // 根据企业id查询年报信息 CompletableFuture<List<CompanyAnnual>> annualFuture = CompletableFuture.supplyAsync(() -> iCompanyAnnualService.listByCompanyIds(ids), asyncTaskExecutor); // 根据查询的年报, 再查询年报子表信息 CompletableFuture<Map<String, String>> annualMapFuture = annualFuture.thenApply(this::getCompanyAnnual); // 得到所有行业 数据小, 直接全查 CompletableFuture<List<CompanyCategoryCode>> categoryCodeFuture = CompletableFuture.supplyAsync(() -> iCompanyCategoryCodeService.list(), asyncTaskExecutor); // 根据企业id查询所属行业代码 CompletableFuture<List<CompanyCategory>> categoryFuture = CompletableFuture.supplyAsync(() -> iCompanyCategoryService.listByCompanyIds(ids), asyncTaskExecutor); 省略.... // 并行执行 CompletableFuture<Void> allOf = CompletableFuture.allOf(annualFuture, categoryCodeFuture, categoryFuture); try { allOf.get(); // 得到所有行业 数据小, 直接全查 List<CompanyCategoryCode> categoryCodeList = categoryCodeFuture.get(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } ``` # 数据库 创建索引 ``` CREATE INDEX indexName ON table_name (column_name) ``` 修改索引 先删除同名索引, 再创建 ``` DROP INDEX [indexName] ON table_name; ``` 数据库 只用来查询, 索引可以随便加 由于查询返回的数据都只有不到8个字段, 大部分是小于3个, 则能添加覆盖索引的就是以覆盖索引.