Quarkus中io.vertx.mutiny多条记录查询流程中断问题求助

1. 确认Multi流的完成信号是否正确发送 #

如果是手动构建Multi流，必须确保调用完成方法告知流结束，否则框架会一直等待最终信号，超时后取消请求：

// 错误示例：未触发完成信号
Multi.createFrom().emitter(emitter -> {
    productList.forEach(emitter::emit);
    // 缺少 emitter.complete();
});

// 正确写法
Multi.createFrom().emitter(emitter -> {
    productList.forEach(emitter::emit);
    emitter.complete(); // 必须调用，标记流结束
});

如果是从数据库查询（比如Panache Reactive），确保使用正确的流式方法，比如findAll().stream()或streamAll()，这类方法会自动处理流的完成信号。

2. 检查REST接口的返回逻辑 #

Quarkus Reactive REST会自动订阅并处理Multi返回值，禁止手动订阅后返回空类型，否则流会因失去框架接管而被取消：

// 错误示例：手动订阅但未返回流
@GET
@Path("/products")
public void getProducts() {
    productService.listProducts().subscribe().with(
        product -> LOG.info(product.getName()),
        failure -> LOG.error(failure)
    );
}

// 正确写法：直接返回Multi让框架处理
@GET
@Path("/products")
public Multi<Product> getProducts() {
    return productService.listProducts();
}

3. 排查数据转换环节的信号传递 #

虽然你已确认“TEST A”日志执行，但要检查转换操作是否阻塞了流或中断了信号传递：

• 如果转换中包含阻塞IO操作，必须切换到Worker线程池执行，避免阻塞事件循环：

return productRepository.findAll()
    .map(productDto -> {
        // 处理阻塞操作（如同步外部API调用）
        return blockingDataConvert(productDto);
    })
    .runSubscriptionOn(Infrastructure.getDefaultWorkerPool()); // 切换线程池

• 使用flatMap()时，确保返回的Uni/Multi能正常完成，未捕获的异常会导致流静默终止，需添加异常处理：

return productRepository.findAll()
    .flatMap(productDto -> {
        return dataConvertService.convert(productDto)
            .onFailure().recoverWithItem(error -> {
                LOG.error("转换失败", error);
                return null; // 或返回默认值，避免流中断
            });
    });

4. 调整请求超时配置 #

默认超时时间过短可能导致长查询被取消，在application.properties中修改相关配置：

# REST接口超时
quarkus.rest-client.read-timeout=30s
# Reactive数据库查询超时
quarkus.hibernate-reactive.query.timeout=30s
# 事件循环线程池配置（如果流处理耗时较长）
quarkus.vertx.event-loops.pool-size=16

5. 确保上下文正确传播 #

如果流处理中依赖请求范围的Bean，需添加上下文捕获，避免上下文丢失导致流异常：

return productService.listProducts()
    .withContextCapture(); // 保留请求上下文

内容的提问来源于stack exchange，提问作者Takayuki-T