java-CompletionService使用小记

由于项目性能需求，之前单线程阻塞查询云端数据效率太慢，需要改进。百度了下CompletionService还是蛮符合需求的。

package com.demo.thread.executor;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecuteMain {

//线程池大小
private static final int THREAD_POOL_SIZE = Runtime.getRuntime().availableProcessors();
//最大线程
private static final int THREAD_MAXIMUM_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2 + 1;

public static void main(String[] args) {
try {
ExecuteMain main = new ExecuteMain();
main.execService();
} catch(InterruptedException e) {
System.out.println("进入异常");
} catch (Exception e) {
e.printStackTrace();
}
}

//执行线程
public void execService() throws Exception {
ExecutorService executorService = new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_MAXIMUM_POOL_SIZE, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(THREAD_MAXIMUM_POOL_SIZE ), new RejectedExecutionHandler() {

@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
try {
//改为阻塞方法
executor.getQueue().put(r);
} catch (Exception e) {

}
}
});
CompletionService<Integer> completionService  = new ExecutorCompletionService<Integer>(executorService);
try {
List<Integer> accountIdList = genAccountId();
for (Integer accountId : accountIdList) {
final finalAccountId = accountId;
completionService.submit(buildTask(accountId));
}

int resultNum = 0;
for (int i = 0; i < accountIdList.size(); i++) {
resultNum = completionService.take().get();
System.out.println("第" +(i + 1)+"个执行结果" + resultNum);
}
System.out.println("全部任务执行完成");
} catch(InterruptedException e) {
e.printStackTrace();
} catch (Exception e) {
executorService.shutdownNow();
throw e;
} finally {
executorService.shutdownNow();
}
}

//生成任务
public Callable<Integer> buildTask(final int accountId) throws InterruptedException {
final Random rand = new Random();
Callable<Integer> task = new Callable<Integer>() {
@Override
public Integer call() throws Exception {
System.out.println("执行任务" + Thread.currentThread().getName());
int time = rand.nextInt(100)*100;
Thread.currentThread().sleep(time);
if(accountId > 10) {
throw new InterruptedException("throws Exception");
}
return accountId;
}
};
return task;
}

//生成账号
public List<Integer> genAccountId() {
List<Integer> accountIdList = new ArrayList<Integer>();
accountIdList.add(3);
accountIdList.add(1);
accountIdList.add(2);
accountIdList.add(4);
accountIdList.add(5);
accountIdList.add(6);
accountIdList.add(7);
accountIdList.add(8);
accountIdList.add(9);
accountIdList.add(10);
accountIdList.add(11);
accountIdList.add(12);
accountIdList.add(13);
accountIdList.add(14);
accountIdList.add(15);
accountIdList.add(16);
accountIdList.add(17);
accountIdList.add(18);
return accountIdList;
}
}

demo如上，结果如下



关于CompletionService的原理，可以先百度了解下。这里写下注意事项

1.completionService.take().get(); 这里是没有顺序的，如果程序执行时对有顺序有要求，需要谨慎

2.当callable中出现异常，外层方法必须调用completionService.take().get();这句才能捕获队列的受检查异常。

3.如果你callable中的方法A，又调用了另外一个方法B，而且B方法存在回调行为（比如说查询出结果后还会做其他处理），这时不要使用shutdownNow()进行关闭。否则可能会出现，云端任务执行状态和本地状态不匹配的情况，因为当出现异常时，你调用shutdownNow()后线程结束了，回调时本地状态未更新。

4.线程任务中的参数一定要用final修饰，不然可能会出现入参和结果不匹配情况