Java实现多线程并发发送携带不同授权头的API请求

核心改造思路 #

原有串行逻辑的问题是单线程循环逐个发起请求，IO等待时间长，改造时不需要调整已有的用户token生成逻辑，只需要把后续的API请求逻辑封装成独立任务提交到线程池执行，同时做好线程隔离、异常处理、任务等待即可。
需要注意几个关键原则：

• 每个请求独立持有自己的授权token，避免多线程共享请求实例导致的授权头串号
• 单独捕获每个请求的异常，避免单个请求失败中断全部任务
• 增加任务等待机制，保证方法返回时所有请求都已处理完成
• 复用线程池，避免频繁创建销毁线程带来的性能开销

改造后代码实现 #

首先引入JUC包下的并发工具类，改造请求发送方法：

import java.util.*;
import java.util.concurrent.*;
//  ThreadFactoryBuilder如果没有引入Guava依赖，可以自行实现ThreadFactory接口，不需要强依赖第三方包
import com.google.common.util.concurrent.ThreadFactoryBuilder;

// 全局复用的IO密集型线程池，线程数根据实际接口压测调整即可
private static final ExecutorService API_REQUEST_POOL = new ThreadPoolExecutor(
        20,
        20,
        60L,
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(100),
        new ThreadFactoryBuilder().setNameFormat("lab-api-request-%d").build(),
        new ThreadPoolExecutor.CallerRunsPolicy()
);

public void sendMultipleAPIRequests() throws InterruptedException {
    if (usersMap == null || usersMap.isEmpty()) {
        throw new IllegalStateException("请先执行createUsers方法初始化用户token映射");
    }

    CountDownLatch finishLatch = new CountDownLatch(usersMap.size());
    // 需要统一收集所有接口返回结果的话，用线程安全的List存储
    List<JsonObject> apiResults = Collections.synchronizedList(new ArrayList<>());

    for (Map.Entry<String, String> entry : usersMap.entrySet()) {
        String currentUser = entry.getKey();
        String currentToken = entry.getValue();
        
        API_REQUEST_POOL.submit(() -> {
            try {
                // 每个任务独立初始化请求客户端，天然隔离授权信息，不会出现token串号问题
                LabAPIChecker checker = new LabAPIChecker(HOST, currentToken);
                JsonObject result = checker.launchLabGetRequest();
                // 此处补充自定义的结果处理逻辑
                System.out.printf("用户[%s]请求完成，返回：%s%n", currentUser, result);
                apiResults.add(result);
            } catch (Exception e) {
                System.err.printf("用户[%s]请求失败，错误：%s%n", currentUser, e.getMessage());
            } finally {
                finishLatch.countDown();
            }
        });
    }

    // 阻塞等待所有请求完成，超时时间根据接口平均响应时间调整
    boolean allDone = finishLatch.await(60, TimeUnit.SECONDS);
    if (!allDone) {
        System.err.println("警告：部分请求在1分钟超时时间内未执行完成");
    }

    // 一次性任务场景下，所有请求跑完可以关闭线程池；如果是频繁调用的方法，保持线程池复用不要关闭
    // API_REQUEST_POOL.shutdown();
}

可选优化点 #

• 如果createUsers方法执行完成后不存在多线程写入usersMap的场景，原有HashMap可以直接使用，不需要替换；如果后续存在多线程读写map的需求，把初始化代码改成usersMap = new ConcurrentHashMap<>()即可
• 如果确认LabAPIChecker底层的HTTP客户端是线程安全的，可以把客户端实例做成全局共享，只在请求时传入当前token，减少对象创建开销；如果不确定线程安全性，保持每个任务新建实例的写法即可，不会出现业务逻辑错误
• 原有createUsers的用户创建、token生成逻辑不建议改成多线程，这类写接口一般有频控限制，串行执行稳定性更高，等所有token生成完成后再批量并发发GET请求即可
• 线程池参数不用硬卡20个线程，根据实际的接口响应时间、服务端限流阈值调整就行，40个请求的场景开10-30个线程都属于合理范围

内容的提问来源于stack exchange，提问作者tupac shakur