正确地使用CompletableFuture

@Test
public void test() throws InterruptedException {
    ThreadPoolExecutor executor = ThreadPoolExecutorBuilder.build(2, 4, 100);
    List<CompletableFuture<?>> fs = new ArrayList<>(size);
    List<CompletableFuture<?>> others = new ArrayList<>();
    for (int i = 0; i < size; i++) {
        MyRandomSleepTask task = new MyRandomSleepTask(ai.incrementAndGet());
        CompletableFuture<?> cf = CompletableFuture.supplyAsync(() -> {
            task.run();
            return null;
        }, executor);
        if (i % 2 == 0) {
            System.out.println("pick task-" + (i + 1));
            fs.add(cf);
        } else {
            others.add(cf);
        }
    }

    CompletableFuture<?>[] fa = new CompletableFuture[0];
    try {
        CompletableFuture.allOf(fs.toArray(fa)).get(2500, TimeUnit.MILLISECONDS);
    } catch (InterruptedException | ExecutionException | TimeoutException e) {
        e.printStackTrace();
    }
    CompletableFuture.allOf(others.toArray(fa)).join();
}

try {
    CompletableFuture.allOf(fs.toArray(fa)).get(2500, TimeUnit.MILLISECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
    e.printStackTrace();
    if (e instanceof TimeoutException) {
        fs.forEach(f -> f.cancel(true));
    }
}

private static final Cache<String, String> cache = CacheBuilder.newBuilder()
        .maximumSize(10_000)
        .expireAfterWrite(3, TimeUnit.SECONDS)
        .expireAfterAccess(3, TimeUnit.SECONDS)
        .weakKeys()
        .weakValues()
        .build();

public class HttpRequestTask implements Callable<String> {
    @Override
    public String call() {
        try {
            if (Thread.currentThread().isInterrupted()) {
                System.out.printf("[" + Thread.currentThread().getName() + "]" + this.getClass().getSimpleName() + "-" + this.id + " is interrupted");
                return null;
            }
            long left = score % 600;
            try {
                System.out.println("[" + Thread.currentThread().getName() + "]" + this.getClass().getSimpleName() + "-" + this.id + " will Sleep(" + left + "ms)...");
                TimeUnit.MILLISECONDS.sleep(left);
            } catch (InterruptedException e) {
                //ignore
                //Thread.currentThread().interrupt();
            }
            String result = cache.getIfPresent(query);
            if (result != null) {
                System.err.println("get result from cache: key=" + query);
                return result;
            }
            String requestUrl = "https://cn.bing.com/search?q=" + query;
            System.out.println("[" + Thread.currentThread().getName() + "]" + this.getClass().getSimpleName() + "-" + this.id + " request (" + requestUrl + ")...");
            result = doRequest(requestUrl);
            if (result != null) {
                cache.put(query, result);
            }
            return result;
        } catch (Exception e) {
        }
        return null;
    }
    protected String doRequest(String requestUrl) throws Exception {
        return MyHttpClient.get(requestUrl);
    }    
}

for (int i = 0; i < size; i++) {
    HttpRequestTask task = new HttpRequestTask(ai.incrementAndGet(), randomChar());
    CompletableFuture<?> cf = CompletableFuture.supplyAsync(() -> {
        return task.call();
    }, executor);
    if (i % 2 == 0) {
        System.out.println("pick task-" + (i + 1));
        fs.add(cf);
    } else {
        others.add(cf);
    }
}

public class AsyncHttpRequestTask extends HttpRequestTask {

    private static AsyncHttpClient asyncHttpClient = asyncHttpClient(Dsl.config().setConnectTimeout(2000).setReadTimeout(2000));

    public AsyncHttpRequestTask(int id, String query) {
        super(id, query);
    }

    @Override
    protected String doRequest(String requestUrl) throws Exception {
        Future<Response> whenResponse = asyncHttpClient.prepareGet("http://www.example.com/").execute();
        Response response = whenResponse.get();
        if (response != null) {
            return response.getResponseBody(StandardCharsets.UTF_8);
        }
        return null;
    }

    public static void closeClient() {
        try {
            asyncHttpClient.close();
        } catch (IOException e) {
            //
        }
    }
}

for (int i = 0; i < size; i++) {
    AsyncHttpRequestTask task = new AsyncHttpRequestTask(ai.incrementAndGet(), randomChar());
    CompletableFuture<?> cf = CompletableFuture.supplyAsync(() -> {
        return task.call();
    }, executor);
    if (i % 2 == 0) {
        System.out.println("pick task-" + (i + 1));
        fs.add(cf);
    } else {
        others.add(cf);
    }
}