Java批量并发请求的三种写法与性能优化

static void sleepseconds(long seconds) {
    try {
        timeunit.seconds.sleep(seconds);
    } catch (interruptedexception e) {
        // 不要吞掉中断信号：恢复中断标记，让上层有机会感知取消
        thread.currentthread().interrupt();
        throw new illegalstateexception("task interrupted", e);
    }
}

static list<callable<integer>> buildtasks() {
    return arrays.aslist(
            () -> { sleepseconds(3); system.out.println("task-3s done"); return 3; },
            () -> { sleepseconds(1); system.out.println("task-1s done"); return 1; },
            () -> { sleepseconds(2); system.out.println("task-2s done"); return 2; }
    );
}

public static void byinvokeall() throws interruptedexception {
    long start = system.currenttimemillis();
    executorservice pool = executors.newfixedthreadpool(3);   // 仅为示例，生产别这么写，见 6.1
    try {
        list<future<integer>> futures = pool.invokeall(buildtasks());
        for (future<integer> f : futures) {
            try {
                system.out.println("result: " + f.get());
            } catch (executionexception e) {
                system.out.println("task failed: " + e.getcause());
            }
        }
    } finally {
        pool.shutdown();
    }
    system.out.println("cost=" + (system.currenttimemillis() - start) + "ms");
}

// 反面示例：忙轮询 isdone()
while (!futures.isempty()) {
    futures.removeif(f -> {
        if (f.isdone()) { consume(f); return true; }
        return false;
    });
}

// 3 秒整体预算，到点未完成的任务会被 cancel
list<future<integer>> futures = pool.invokeall(buildtasks(), 3, timeunit.seconds);
for (future<integer> f : futures) {
    if (f.iscancelled()) {
        system.out.println("timeout, skipped");     // 超时被取消
        continue;
    }
    try {
        system.out.println("result: " + f.get());
    } catch (executionexception e) {
        system.out.println("failed: " + e.getcause());
    }
}

public static void bycompletionservice() {
    long start = system.currenttimemillis();
    executorservice pool = executors.newfixedthreadpool(3);
    try {
        list<callable<integer>> tasks = buildtasks();
        // 注意：completionservice 要每批次新建，不要做成共享单例，见 3.4
        completionservice<integer> cs = new executorcompletionservice<>(pool);
        tasks.foreach(cs::submit);

        for (int i = 0; i < tasks.size(); i++) {
            try {
                integer r = cs.take().get();     // 按完成顺序返回
                system.out.println("result: " + r);
            } catch (executionexception e) {
                system.out.println("task failed: " + e.getcause());
            } catch (interruptedexception e) {
                thread.currentthread().interrupt();
                break;
            }
        }
    } finally {
        pool.shutdown();
    }
    system.out.println("cost=" + (system.currenttimemillis() - start) + "ms");
}

public class batchfanout {

    public static class batchresult<t> {
        public final list<t> succeeded = new arraylist<>();
        public int failed;
        public int unfinished;
        public boolean ispartial() { return failed > 0 || unfinished > 0; }
    }

    /**
     * 在 budgetms 的整体预算内并发执行 tasks，返回已完成的结果，未完成的取消。
     */
    public static <t> batchresult<t> execute(list<callable<t>> tasks,
                                             executorservice pool,
                                             long budgetms) {
        batchresult<t> result = new batchresult<>();
        completionservice<t> cs = new executorcompletionservice<>(pool);

        list<future<t>> submitted = new arraylist<>(tasks.size());
        for (callable<t> task : tasks) {
            submitted.add(cs.submit(task));
        }

        long deadline = system.nanotime() + timeunit.milliseconds.tonanos(budgetms);
        try {
            for (int i = 0; i < tasks.size(); i++) {
                long remaining = deadline - system.nanotime();
                future<t> f = remaining <= 0 ? null : cs.poll(remaining, timeunit.nanoseconds);
                if (f == null) {                       // 预算用尽
                    result.unfinished = tasks.size() - i;
                    break;
                }
                try {
                    result.succeeded.add(f.get());     // 此处不会再阻塞
                } catch (executionexception e) {
                    result.failed++;
                    // 生产代码请打日志并带上任务标识，见第七章
                }
            }
        } catch (interruptedexception e) {
            thread.currentthread().interrupt();
        } finally {
            // 必须收尾：提前 break 后，未完成任务仍在占用线程池
            submitted.foreach(f -> f.cancel(true));
        }
        return result;
    }
}

public static void bycompletablefuture(executorservice pool) {
    long start = system.currenttimemillis();

    list<integer> seconds = arrays.aslist(3, 1, 2);
    list<completablefuture<integer>> futures = seconds.stream()
            .map(s -> completablefuture.supplyasync(() -> {
                sleepseconds(s);
                system.out.println("task-" + s + "s done on " + thread.currentthread().getname());
                return s;
            }, pool))                                   // 显式指定线程池，见 4.2
            .collect(collectors.tolist());

    // 完成即消费（谁先完成谁先进来）
    futures.foreach(f -> f.thenaccept(r -> system.out.println("result: " + r)));

    // 等全部结束；任一异常会在这里抛 completionexception
    completablefuture.allof(futures.toarray(new completablefuture[0])).join();

    system.out.println("cost=" + (system.currenttimemillis() - start) + "ms");
}

threadpoolexecutor batchpool = new threadpoolexecutor(
        8, 8,
        60l, timeunit.seconds,
        new arrayblockingqueue<>(200),                              // 有界
        new customizablethreadfactory("batch-fetch-"),              // spring 提供，线程名可辨识
        new threadpoolexecutor.abortpolicy());                      // 快速失败优于悄悄堆积
batchpool.allowcorethreadtimeout(true);

completablefuture.supplyasync(() -> client.fetchpage(pageno), batchpool)
        .thenaccept(this::consume)                 // 成功路径
        .exceptionally(ex -> {                     // 失败路径，返回兜底值让链继续
            log.warn("fetch page {} failed", pageno, ex);
            return null;
        });

completablefuture<list<pageresult>> all = completablefuture
        .allof(futures.toarray(new completablefuture[0]))
        .thenapply(v -> futures.stream()
                .map(completablefuture::join)      // 此时都已完成，join 不阻塞
                .collect(collectors.tolist()));

completablefuture<object> fastest = completablefuture.anyof(fromcache, fromdb);

completablefuture<pageresult> f = completablefuture
        .supplyasync(() -> client.fetchpage(pageno), batchpool)
        .ortimeout(500, timeunit.milliseconds)                   // 超时则异常完成
        .exceptionally(ex -> pageresult.empty(pageno));           // 降级为空页

// 或者直接给默认值，不走异常路径
completablefuture<pageresult> f2 = completablefuture
        .supplyasync(() -> client.fetchpage(pageno), batchpool)
        .completeontimeout(pageresult.empty(pageno), 500, timeunit.milliseconds);

// spring 项目
new customizablethreadfactory("order-batch-fetch-");

// 或用 jdk 原生方式
threadfactory factory = r -> {
    thread t = new thread(r, "order-batch-fetch-" + counter.incrementandget());
    t.setdaemon(false);
    t.setuncaughtexceptionhandler((th, e) -> log.error("uncaught in {}", th.getname(), e));
    return t;
};

public class mdctaskdecorator implements taskdecorator {
    @override
    public runnable decorate(runnable runnable) {
        map<string, string> parent = mdc.getcopyofcontextmap();   // 提交线程的上下文
        return () -> {
            try {
                if (parent != null) {
                    mdc.setcontextmap(parent);
                }
                runnable.run();
            } finally {
                mdc.clear();                                       // 池化线程必须清理，防止污染下一个任务
            }
        };
    }
}

@bean("batchexecutor")
public threadpooltaskexecutor batchexecutor() {
    threadpooltaskexecutor executor = new threadpooltaskexecutor();
    executor.setcorepoolsize(8);
    executor.setmaxpoolsize(8);
    executor.setqueuecapacity(200);
    executor.setthreadnameprefix("batch-fetch-");
    executor.settaskdecorator(new mdctaskdecorator());
    executor.initialize();
    return executor;
}

executorservice monitored = executorservicemetrics.monitor(
        meterregistry, batchpool, "batch-fetch", tags.of("biz", "order"));