在处理大数据或实时数据时，如果在主线程频繁创建大量对象，这些对象使用完后成为游离对象，不会立即被GC。当创建速度大于销毁速度时，可能导致内存持续上涨，最后内存溢出。
可以开启多线程来处理，线程内的对象会在执行结束后尽快的销毁，均分内存累加的负担，保证内存占用的稳定性。
springboot的多线程使用
- 配置@EnableAsync
package com.cdgs.data.config;
import java.util.concurrent.Executor;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.aop.interceptor.SimpleAsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@Configuration
@EnableAsync(proxyTargetClass=true)//利用@EnableAsync注解开启异步任务支持
@ComponentScan("com.cdgs.data.service") //必须加此注解扫描包
public class CustomMultiThreadingConfig implements AsyncConfigurer{
@Override
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(10);
taskExecutor.setMaxPoolSize(20);
taskExecutor.setQueueCapacity(500);
//当提交的任务个数大于QueueCapacity，就需要设置该参数，但spring提供的都不太满足业务场景，可以自定义一个，也可以注意不要超过QueueCapacity即可
//taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
taskExecutor.setAwaitTerminationSeconds(10);
taskExecutor.setThreadNamePrefix("ES-IMOPRT-");
taskExecutor.initialize();
return taskExecutor;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return new SimpleAsyncUncaughtExceptionHandler();
}
}
- 配置@Async，修饰类时表示类里所有方法都是多线程异步执行
@Async
public Future<Integer> jdbcToElasticsearch(Pageable pageable) {
//通过实现ApplicationContextAware得到applicationContext，进而获取spring管理的bean
BaseinfoRepository repository = (BaseinfoRepository)SpringBeanUtil.getBean(BaseinfoRepository.class);
List<ZrHisEnterpriseBaseinfo> list = new ArrayList<>() ;
//实际查询数据库等业务代码...
elasticsearchOperation.bulkSave(list);
return new AsyncResult<>(list.size());
}
- 调用，注意：调用方法不能和异步方法在同一类里
//oralce批量导入es
@GetMapping("/baseinfo")
public String importBaseinfo(@PageableDefault(size=1000)Pageable pageable) {
long count = autoImportService.count();
int size = pageable.getPageSize();
long loops = count%size>0?count/size+1:count/size;
ArrayList<Future<Integer>> futureList = new ArrayList<>();
for (int i = 0; i < loops; i++) {
//异步执行任务，返回参数使用Future封装接收
Future<Integer> future = autoImportService.jdbcToElasticsearch(new PageRequest(pageable.getPageNumber() + i, size));
futureList.add(future);
}
importCount = checkTaskDone(futureList);
System.out.println(importCount);
}
public static long checkTaskDone(ArrayList<Future<Integer>> futureList) {
//判断异步调用的方法是否全都执行完了
while(true) {
int doneSize = 0;
for ( Future<Integer> future : futureList) {
//该异步方法是否执行完成
if(future.isDone()) {
doneSize++;
}
}
//如果异步方法全部执行完，跳出循环
if(doneSize == futureList.size()) {
break;
}
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}//每隔2秒判断一次
}
long importCount = 0;
for ( Future<Integer> future : futureList) {
try {
importCount += future.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
return importCount;
}
调用该方法，观察内存能够稳定在一定范围
QQ图片20190304090025.png
