FixedThreadPool设为7时任务耗时异常飙升问题排查

问题现象 #

• 用8个工作线程处理8个OpenAI调用任务时，总耗时约10秒
• 将基于虚拟线程的FixedThreadPool核心线程数设为7后，处理8个任务总耗时接近70秒：前7个任务10秒完成，最后一个任务等待60秒才开始执行

核心代码片段 #

TaskThread 实现 #

private final ExecutorService executorService;
private final OpenAiService openAiService;
private final List<String> workerJobs;

public TaskThread(List<String> workerJobs, OpenAiService openAiService) {
    this.workerJobs = workerJobs;
    this.openAiService = openAiService;
    this.executorService = Executors.newFixedThreadPool(7, Thread.ofVirtual().name("Thread-", 1).factory());
}

@Override
public void run() {
    List<CompletableFuture<String>> jobs = workerJobs.stream()
        .map(model -> CompletableFuture.supplyAsync(new WorkerJob(openAiService, model), executorService))
        .toList();

    List<String> taskResultModels = jobs
        .stream()
        .map(CompletableFuture::join)
        .toList();
    
    // 保存结果到数据库
}

WorkerJob 实现 #

public class WorkerJob implements Supplier<String> {

    private final OpenAiService openAiService;
    private final String model;

    public WorkerJob(OpenAiService openAiService, String model) {
        this.openAiService = openAiService;
        this.model = model;
    }

    @Override
    public String get() {
        try {
            return openAiService.generate(model);
        } catch (Exception e) {
            return "Error";
        }
    }
}

OpenAiService 实现 #

@Service
class OpenAiService {
    private final OpenAIClient openAIClient;

    @Autowired
    public OpenAiService(OpenAIClient openAIClient) {
        this.openAIClient = openAIClient;
    }

    public String generate(String model) {
        try {
            String content = null;
            List<ChatRequestMessage> chatMessages = new ArrayList<>();
            chatMessages.add(new ChatRequestUserMessage(model));
            ChatCompletions chatCompletions = openAIClient.getChatCompletions("id", new ChatCompletionsOptions(chatMessages));
            for (ChatChoice choice : chatCompletions.getChoices()) {
                ChatResponseMessage message = choice.getMessage();
                content = message.getContent();
            }

            return content;
        } catch (Exception e) {
            return "Error";
        }
    }
}

TaskThread 启动方式 #

Executors.newFixedThreadPool(2).submit(new TaskThread(List.of(""), openAiService));

问题根因 #

1. 虚拟线程与同步调用不匹配：虚拟线程优势是处理大量IO阻塞任务，但OpenAIClient.getChatCompletions是同步阻塞调用，且客户端内部可能存在连接池上限，7个并发调用直接占满连接池，第8个请求被客户端内部排队，而非线程池排队
2. 服务端限流：Azure OpenAI服务端对单账号并发请求有限制，7个并发已达阈值，第8个请求被服务端限流排队，导致等待时间大幅拉长——即使单独创建客户端，服务端限流规则依然生效

解决方案 #

方案1：调整线程池与客户端配置 #

• 增大OpenAIClient的连接池最大连接数，避免客户端内部排队
• 将线程池调整为足够大的虚拟线程池（比如16），让线程池负责任务排队，而非客户端或服务端

方案2：添加并发限流控制 #

在OpenAiService中用Semaphore控制并发请求数，匹配服务端允许的阈值：

@Service
class OpenAiService {
    private final OpenAIClient openAIClient;
    private final Semaphore semaphore;

    @Autowired
    public OpenAiService(OpenAIClient openAIClient) {
        this.openAIClient = openAIClient;
        // 设置为服务端允许的最大并发数，示例值为7
        this.semaphore = new Semaphore(7);
    }

    public String generate(String model) {
        try {
            semaphore.acquire();
            String content = null;
            List<ChatRequestMessage> chatMessages = new ArrayList<>();
            chatMessages.add(new ChatRequestUserMessage(model));
            ChatCompletions chatCompletions = openAIClient.getChatCompletions("id", new ChatCompletionsOptions(chatMessages));
            for (ChatChoice choice : chatCompletions.getChoices()) {
                ChatResponseMessage message = choice.getMessage();
                content = message.getContent();
            }
            return content;
        } catch (Exception e) {
            return "Error";
        } finally {
            semaphore.release();
        }
    }
}

方案3：改用异步客户端 #

使用Azure OpenAI的异步客户端（OpenAIClientAsync），结合虚拟线程实现非阻塞调用，让少量虚拟线程即可处理大量并发IO任务，从根本上避免排队阻塞问题。

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