本文介绍一种基于 concurrenthashmap + future 的轻量级方案，确保多个线程对同一 requestid 的任务串行执行，不同 requestid 则可并行处理，兼顾性能与正确性。
本文介绍一种基于 concurrenthashmap + future 的轻量级方案，确保多个线程对同一 requestid 的任务串行执行，不同 requestid 则可并行处理，兼顾性能与正确性。
在高并发请求处理场景中，常需保证「相同业务标识（如 RequestID）的任务不被并发执行」，而不同标识之间应尽可能并行以提升吞吐。Long 类型的 RequestID 作为自然键，天然适合用作同步粒度的依据。直接使用全局锁（如 synchronized）会严重降低并发度；而为每个可能的 ID 预分配锁对象又存在内存与管理开销。理想方案应满足：按需创建、无内存泄漏、自动清理、线程安全、低侵入。
以下实现采用 ConcurrentHashMap<long future>></long> 作为「进行中请求注册表」，核心思想是：
- 每次提交任务前，以
requestId为 key 调用compute()方法； - 若该 ID 当前无活跃任务（即
existingFuture == null），则提交新任务并阻塞等待结果； - 若已有任务正在执行（
existingFuture != null），则当前调用将等待已有 Future 完成后复用其结果（本示例简化为拒绝重入，实际可扩展为排队或返回 pending 状态）； -
compute()的原子性保证了「检查-注册-提交」三步不可分割，彻底避免竞态。
关键代码逻辑如下（精简自原始示例）：
private static final ConcurrentHashMap<Long, Future<?>> inProgressRequestIds = new ConcurrentHashMap<>();
public T runATaskForId(Callable<T> task, long requestId) {
return processRequest(new RequestTask(requestId, task));
}
private T processRequest(RequestTask task) {
inProgressRequestIds.compute(task.getRequestId(), (key, existing) -> {
// 仅当无现存任务时才提交新任务
if (existing == null || existing.isDone() || existing.isCancelled()) {
Future<T> future = executor.submit(task);
try {
task.setResult(future.get()); // 同步阻塞获取结果
} catch (Exception e) {
throw new RequestProcessingException("Failed to process request " + key, e);
}
} else {
// 可选：等待已有 Future 或抛出 BusyException
try {
task.setResult((T) existing.get());
} catch (Exception e) {
throw new RequestProcessingException("Failed to wait for existing request " + key, e);
}
}
return null; // 清除映射，避免长期持有引用（重要！）
});
return task.getResult();
}⚠️ 注意事项：
-
compute()中返回null是关键——它主动从ConcurrentHashMap中移除条目，防止内存泄漏（尤其当 RequestID 空间无限时）； -
Future.get()会阻塞当前线程，若需异步响应，应改用CompletableFuture链式编排； - 生产环境建议增加超时机制（如
future.get(30, TimeUnit.SECONDS)）并妥善处理TimeoutException； -
ExecutorService生命周期需由上层统一管理（如 Spring Bean 的@PreDestroy或显式shutdown()），避免线程池泄露； - 若 RequestID 可能为
null，须提前校验，否则ConcurrentHashMap抛NullPointerException。
该方案以极小的同步开销实现了细粒度请求级串行化，在电商订单处理、金融交易幂等执行、分布式任务调度等场景中具备良好适用性。
