# 需求: 公司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個, 則能添加覆蓋索引的就是以覆蓋索引.