情景：1000万表数据导入内存数据库，按分页大小10000查询，多线程，15条线程跑。
使用了ExecutorService executor = Executors.newFixedThreadPool(15)
本地跑了一段时间后，发现电脑CPU逐渐升高，最后CPU占用100%卡死，内存使用也高达80%。
Debug 发现虽然创建了定长15的线程池，但是因为数据量大，在For中循环分页查询的List会持续加入LinkedBlockingQueue
队列中每一个等待的任务，又加载了1万的数据。所以不管是线程数的CPU抢占，还是内存的消耗都是极高。
所以是不是能够控制等待队列LinkedBlockingQueue的上限就可以了。
使用AtomicLong 统计线程是否完成，再执行executor.submit()提交新的任务导队列中。
伪代码如下：
private AtomicLong threadNum = new AtomicLong(0);
public void init() throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(15);
Integer total = accountMapper.selectCount(new QueryWrapper<>());
Integer pageSize = 10000; // 页大小
Integer pageCount = (total + pageSize -1) / pageSize; // 总页数
for (Integer start = 1; start <= pageCount; start++) {
List<Account> list = accountMapper.selectPage(new Page<>(start, pageSize), query).getRecords();
//等待线程任务完成，设置30，可令运行线程数为15，等待队列线程数为15
while (threadNum.get() >= 30){
Thread.sleep(5000);
}
//开启1个线程+1
threadNum.incrementAndGet();
executor.submit(() -> {
try {
// 处理业务
dealMessage(list);
// 任务完成 -1
threadNum.decrementAndGet();
} catch (Exception e) {
e.printStackTrace();
}
});
}
executor.shutdown();
executor.awaitTermination(1, TimeUnit.DAYS);
}
效果就是CPU保持在15~45%之间，内存占用也只有45%。
目前只想到这样的方式，控制等待队列LinkedBlockingQueue的上限，还有更好的方式请告知，感谢！
2021-02-03-分割线
最近又用到了多线程开发，发现了还是有很多方式控制的。简单的使用java的Semaphore令牌限流控制也能实现。
public static void main(String[] args) {
final List<String> tableNames = new ArrayList<>();
tableNames.add("a");
tableNames.add("b");
tableNames.add("c");
tableNames.add("d");
tableNames.add("e");
tableNames.add("f");
final Semaphore semaphore = new Semaphore(2);
final ExecutorService fixedThread = Executors.newCachedThreadPool();
for (final String tableName : tableNames) {
//阻塞，获取令牌
try {
semaphore.acquire();
} catch (InterruptedException e) {
e.printStackTrace();
}
//do
fixedThread.execute(() -> { //can throw ex log
final ExecutorService executorService = Executors.newCachedThreadPool();
try {
executorService.submit(() -> { //can't throw ex log
//int i = 1/0;
System.out.println("tableName2:" + tableName);
});
//int i = 1/0;
System.out.println("tableName:" + tableName);
} catch (Exception e) {
e.printStackTrace();
} finally {
executorService.shutdown();
try {
executorService.awaitTermination(1, TimeUnit.DAYS);
} catch (InterruptedException e) {
e.printStackTrace();
}
semaphore.release();
System.out.println("semaphore.release");
}
});
}
// 记得关闭线程池
fixedThread.shutdown();
try {
fixedThread.awaitTermination(1, TimeUnit.DAYS);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("主线程...");
}
打印结果
tableName:b
tableName2:b
tableName:a
tableName2:a
semaphore.release
semaphore.release
tableName:d
tableName2:d
tableName:c
semaphore.release
tableName:e
tableName2:c
semaphore.release
tableName:f
tableName2:e
semaphore.release
tableName2:f
semaphore.release
主线程...
