使用场景:将耗时任务同时执行,比如通过rpc从多接口获取数据组装的时候,可以同时调多个接口。
demo
package a;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class CompletionServiceTest {
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = CompletionServiceTest.exec;
CompletionService completionService = new ExecutorCompletionService(executorService);
List<Future> taskList = new ArrayList<Future>();
CompletionServiceTest completionServiceTest=new CompletionServiceTest();
Datatest datatest=new Datatest();
taskList.add(completionService.submit(() -> {
datatest.a=completionServiceTest.add1();
return true;
}));
taskList.add(completionService.submit(() -> {
datatest.b=completionServiceTest.add2();
return true;
}));
taskList.add(completionService.submit(() -> {
datatest.c=completionServiceTest.add3();
return true;
}));
taskList.add(completionService.submit(() -> {
datatest.d=completionServiceTest.add4();
return true;
}));
Long time1=System.currentTimeMillis();
for (int i = 0, size = taskList.size(); i < size; i++) {
try {
completionService.take().get();
} catch (ExecutionException e) {
throw new RuntimeException("异步查询异常");
}
}
System.out.println(datatest);
System.out.println("异步执行时间:"+(System.currentTimeMillis()-time1));
time1=System.currentTimeMillis();
completionServiceTest.add1();
completionServiceTest.add2();
completionServiceTest.add3();
completionServiceTest.add4();
System.out.println("主线程执行时间:"+(System.currentTimeMillis()-time1));
}
/**
* 模拟耗时的一些操作
*/
public Integer add1() throws InterruptedException {
Thread.sleep(3000);
System.out.println("add1"+Thread.currentThread().getName());
return 1;
}
/**
* 模拟耗时的一些操作
*/
public Integer add2() throws InterruptedException {
Thread.sleep(2000);
System.out.println("add2"+Thread.currentThread().getName());
return 2;
}
/**
* 模拟耗时的一些操作
*/
public Integer add3() throws InterruptedException {
Thread.sleep(800);
System.out.println("add3"+Thread.currentThread().getName());
return 3;
}
/**
* 模拟耗时的一些操作
*/
public Integer add4() throws InterruptedException {
Thread.sleep(600);
System.out.println("add4"+Thread.currentThread().getName());
return 4;
}
/**
* 数据实体
*/
static class Datatest{
Integer a;
Integer b;
Integer c;
Integer d;
@Override
public String toString() {
return "Datatest{" +
"a=" + a +
", b=" + b +
", c=" + c +
", d=" + d +
'}';
}
}
public static final ExecutorService exec = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors() * 50, 0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
new ThreadFactory() {
private final String threadPoolPrefix = "ThreadPool-Thread-";
public AtomicInteger counter = new AtomicInteger(0);
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName(threadPoolPrefix+counter.incrementAndGet());
//设定为守护线程
if (t.isDaemon()){
t.setDaemon(false);
}
//设定线程优先级
if (t.getPriority() != Thread.NORM_PRIORITY){
t.setPriority(Thread.NORM_PRIORITY);
}
return t;
}
});
}
|