基于Quarkus、Mutiny、Kotlin的异步回调处理最佳模式咨询

我需要对外暴露一个同步API，底层调用异步服务。当异步服务侧的数据就绪时，会回调我的服务器；若在合理时间内数据未就绪，API需返回404。

我已实现一个基于CompletableFuture的解决方案，还制作了另一个使用Emitter的原型——将Emitter存储在上下文中，以便收到回调时检索，但不确定这是否是最简最优的实现方式，想听听大家的看法。

@Path("/status")
class DeviceStatusController(
    private val deviceStatusService: DeviceStatusService
) {

    @POST
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    fun getStatus(@Valid @NotNull requestStatus: RequestStatus): Uni<String> {
        return deviceStatusService.getStatus(requestStatus)
    }
}

@ApplicationScoped
@Path("/notifications")
class NotificationController(val service: DeviceStatusService) {

    @POST
    fun receive(@RequestBody @Valid notification: Notification) {
        service.receive(notification)
    }
}

@ApplicationScoped
class DeviceStatusService {
    private val logger: Logger = Logger.getLogger(this.javaClass.canonicalName)

    data class UniContext(val emitter: CompletableFuture<String>)

    //Todo: Change for redis registration with TTL
    val notificationMap = mutableMapOf<String, UniContext>()

    fun getStatus(requestStatus: RequestStatus): Uni<String> {

        // Call async service here ... it will call us back on notifications endpoint
        // It will just reply ... your request is accepted

        val emitter = CompletableFuture<String>()
        notificationMap[requestStatus.device] = UniContext(emitter)
        return Uni.createFrom().completionStage(emitter.minimalCompletionStage())
            .ifNoItem().after(Duration.ofSeconds(60))
            .failWith {
                cleanContexts(requestStatus.device)
                TimeoutException("No reply for ${requestStatus.device}")
            }
    }

    fun cleanContexts(id: String) {
        notificationMap.remove(id)
    }

    fun receive(notification: Notification) {
        logger.fine("Notification received for ${notification.device}")
        notificationMap[notification.device]?.let {
            it.emitter.complete(notification.status)
            notificationMap.remove(notification.device)
        }
    }

}

现有CompletableFuture方案的优缺点 #

优点 #

• 结合Mutiny的Uni与CompletableFuture，贴合Quarkus异步编程模型，逻辑直观易懂
• 内置超时处理，超时后自动清理上下文，避免内存泄漏
• 回调处理逻辑清晰，收到通知后完成Future并移除上下文

潜在问题与优化点 #

1. 线程安全风险：当前用mutableMapOf存储上下文，多并发场景下会出现线程安全问题，建议替换为ConcurrentHashMap
2. 内存泄漏隐患：虽然超时会清理上下文，但如果异步服务永远不回调且超时逻辑失效，会导致条目常驻内存。后续换成带TTL的Redis是正确方向，Redis自动过期机制能从根源解决该问题
3. 异常处理完善性：当前未处理CompletableFuture异常完成的场景（比如异步服务返回错误），建议在onFailure中补充日志或错误映射逻辑
4. 状态码映射：当前超时抛出TimeoutException，需确保Quarkus将其映射为404响应，可通过@ServerExceptionMapper统一处理：

@ServerExceptionMapper
fun mapTimeout(e: TimeoutException): Response {
    return Response.status(Response.Status.NOT_FOUND)
        .entity("请求超时，未获取到设备状态")
        .build()
}

Emitter方案对比 #

用Emitter（如Mutiny的Emitter或JAX-RS的AsyncResponse）的思路和CompletableFuture方案本质一致，都是存储"等待器"等待回调触发响应。两者核心逻辑差异不大：

• Mutiny的Emitter更贴合Quarkus的Reactive生态，若服务整体为Reactive架构，使用Emitter能保持技术栈统一
• CompletableFuture是Java标准API，学习成本更低，对熟悉Java并发的开发者更友好
• 从实现简洁性来看，你的CompletableFuture方案已经足够简洁，无需强行替换为Emitter

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