本文介绍一种基于 CompletableFuture 与磁盘持久化的长轮询数据同步方案，替代传统 SynchronousQueue#take() 的阻塞调用，避免因远程服务异常导致的线程挂起、消息丢失及资源泄漏问题。
本文介绍一种基于 `completablefuture` 与磁盘持久化的长轮询数据同步方案，替代传统 `synchronousqueue#take()` 的阻塞调用，避免因远程服务异常导致的线程挂起、消息丢失及资源泄漏问题。
在基于 HTTP 长轮询的跨服务数据同步场景中，直接使用 SynchronousQueue#take() 存在显著缺陷：该方法会无限期阻塞，一旦下游消费者（如客户端断连、超时或服务崩溃），对应线程将无法被唤醒，已入队但未消费的消息可能被“静默丢弃”，且应用重启后数据完全丢失。
为彻底解决这一问题，推荐采用非阻塞 + 异步完成 + 持久化兜底的设计范式：
✅ 核心设计思想
-
解耦等待与生产：不依赖阻塞队列的 take()，而是用 ConcurrentLinkedQueue
> 管理待响应的消费者承诺； - 消息异步匹配：通过 match() 方法主动检查“待消费消息”与“待响应消费者”队列，一旦两者均非空，立即完成 CompletableFuture 并清理资源；
- 磁盘持久化保障：所有待分发消息序列化至临时文件（如 UUID.json），确保 JVM 崩溃后仍可恢复；
- 生命周期联动：DeferredResult.onCompletion() 绑定 future.cancel(true)，客户端断连时主动中断未完成的 Future，防止内存泄漏。
? 关键代码逻辑说明
// 服务端：维护两个并发安全队列
private final Queue<completablefuture>> consumers = new ConcurrentLinkedQueue();
private final Queue<contentandfile> messages = new ConcurrentLinkedQueue();
public CompletableFuture<longpollingdto> getFutureDto() {
CompletableFuture<longpollingdto> future = new CompletableFuture();
consumers.add(future);
match(); // 立即尝试匹配，避免延迟
return future;
}
public void enqueue(LongPollingDto dto) throws IOException {
File file = new File(syncTmpDataDirectory, UUID.randomUUID() + ".json");
writer.writeValue(file, dto);
messages.add(new ContentAndFile(dto, file));
match();
}
private void match() throws IOException {
while (!consumers.isEmpty() && !messages.isEmpty()) {
CompletableFuture<longpollingdto> future = consumers.poll();
if (future.isCancelled()) continue; // 跳过已取消的请求
ContentAndFile item = messages.poll();
future.complete(item.dto);
Files.deleteIfExists(item.file.toPath()); // 安全删除已消费文件
}
}</longpollingdto></longpollingdto></longpollingdto></contentandfile></completablefuture>
⚠️ 注意事项：
- match() 必须是循环执行（而非单次），因为一次匹配可能释放多个 Future，而新消息/消费者可能在匹配过程中持续入队；
- CompletableFuture.isCancelled() 判断必不可少——onCompletion 触发后 Future 状态变为 CANCELLED，此时不应再 complete()；
- 文件操作需配合 try-catch 与日志，避免单个文件损坏阻塞全局匹配流程；
- ApplicationReadyEvent 中的初始化同步确保服务启动时能加载残留临时文件，实现 crash-recovery。
? 控制器层协同设计
@GetMapping("long-polling")
public DeferredResult<longpollingdto> longPolling() {
DeferredResult<longpollingdto> result = new DeferredResult(30_000L); // 可设超时
CompletableFuture<longpollingdto> future = service.getFutureDto();
executor.execute(() -> {
try {
result.setResult(future.get(30, TimeUnit.SECONDS)); // 建议设 Future 超时
} catch (TimeoutException e) {
result.setErrorResult("Request timeout");
} catch (ExecutionException | InterruptedException e) {
result.setErrorResult("Internal error");
Thread.currentThread().interrupt();
} catch (CancellationException e) {
log.debug("Client disconnected, future cancelled.");
}
});
// 客户端断开或超时后自动取消 Future
result.onCompletion(() -> future.cancel(true));
return result;
}</longpollingdto></longpollingdto></longpollingdto>
此设计将控制权从“被动阻塞等待”转为“主动状态驱动”，既规避了 SynchronousQueue 的线程不可中断缺陷，又通过磁盘持久化+内存队列双保险杜绝数据丢失，同时天然支持水平扩展（多实例可通过共享存储协调，或引入分布式锁优化）。
最终效果：任意时刻最多仅有一个 match() 任务活跃，CPU 开销极低；每个消息严格一对一交付；服务重启无数据损失；客户端异常断连后资源毫秒级回收。
