Quarkus中java.net.SocketTimeoutException问题及手动关流方案咨询

问题核心 #

迭代List<Supplier<Stream<TSRecord>>>处理关联数据源连接时，单个流处理耗时数分钟，导致处理第二个元素时抛出java.net.SocketTimeoutException: Read timed out。仅调高超时无法适配列表长度差异，需要可靠的手动/自动关闭流及连接的方案，原Glassfish的CloseableService在Quarkus中无效，不确定如何利用Quarkus/RestEasy的资源管理机制解决。

可行解决方案 #

1. 确保StreamAndName正确实现AutoCloseable并封装连接关闭逻辑 #

你的代码中已经用try-with-resources管理StreamAndName，但关键是**StreamAndName必须正确实现AutoCloseable接口，且close()方法要负责关闭其关联的底层数据源连接**（比如JDBC的Connection、ResultSet，或者其他客户端连接）。

示例修改StreamAndName：

public class StreamAndName<T> implements AutoCloseable {
    private final String name;
    private final Stream<T> stream;
    private final Closeable underlyingResource; // 比如JDBC Connection、客户端连接

    public StreamAndName(String name, Stream<T> stream, Closeable underlyingResource) {
        this.name = name;
        this.stream = stream;
        this.underlyingResource = underlyingResource;
    }

    // 原有getter方法
    public String name() { return name; }
    public Stream<T> stream() { return stream; }

    @Override
    public void close() throws Exception {
        // 先关闭流
        stream.close();
        // 再关闭底层资源（数据源连接等）
        if (underlyingResource != null) {
            underlyingResource.close();
        }
    }
}

这样在try (final StreamAndName<TSRecord> next = sup.get())结束时，会自动调用close()，确保连接被及时释放，不会因为长时间占用导致后续请求超时。

2. 显式强制关闭流与资源（兜底方案） #

如果StreamAndName的close()逻辑不够可靠，可以在处理完单个流后，主动显式关闭流及关联资源：

@Override
public void write(final OutputStream output) throws IOException, WebApplicationException {
    try (final ZipOutputStream zos = new ZipOutputStream(output)) {
        for (Supplier<StreamAndName<TSRecord>> sup : streamSuppliers) {
            StreamAndName<TSRecord> next = null;
            try {
                next = sup.get();
                LOGGER.debug("Writing entry: {}", next.name());
                final ZipEntry entry = new ZipEntry(next.name().concat(".csv"));
                zos.putNextEntry(entry);
                final CsvStreamingOutput so = new CsvStreamingOutput(parameters, resolution, next.stream());
                so.write(zos);
                // 显式关闭流
                next.stream().close();
            } finally {
                zos.closeEntry();
                // 手动关闭StreamAndName及底层资源
                if (next != null) {
                    try {
                        next.close();
                    } catch (Exception e) {
                        LOGGER.error("Failed to close StreamAndName", e);
                    }
                }
            }
        }
    }
}

3. Quarkus下的资源管理适配 #

• 避免使用QuarkusBuildCloseablesBuildItem：它确实是给扩展开发者用的，不适合业务代码。
• RestEasyContext的用法：如果你的数据源连接是在RestEasy请求上下文中创建的，可以将Closeable资源推入上下文，请求结束时自动关闭：

  // 在创建StreamAndName的地方，将底层连接推入RestEasyContext
  Connection conn = dataSource.getConnection();
  Stream<TSRecord> stream = ...; // 从conn获取流
  StreamAndName<TSRecord> san = new StreamAndName<>("name", stream, conn);
  ResteasyContext.pushContext(Closeable.class, conn);
  

  不过更直接的还是让StreamAndName自己管理关闭逻辑，避免依赖上下文的额外配置。

4. 拆分处理流程，减少连接占用时间 #

如果单个流处理耗时极长，可以先将每个流的数据写入本地临时文件，关闭数据源连接后再统一打包成Zip：

@Override
public void write(final OutputStream output) throws IOException, WebApplicationException {
    List<Path> tempFiles = new ArrayList<>();
    try {
        // 第一步：逐个处理流，写入临时文件，立即关闭连接
        for (Supplier<StreamAndName<TSRecord>> sup : streamSuppliers) {
            try (final StreamAndName<TSRecord> next = sup.get()) {
                Path tempFile = Files.createTempFile(next.name(), ".csv");
                tempFiles.add(tempFile);
                try (FileOutputStream fos = new FileOutputStream(tempFile.toFile())) {
                    final CsvStreamingOutput so = new CsvStreamingOutput(parameters, resolution, next.stream());
                    so.write(fos);
                }
            }
        }
        // 第二步：将临时文件打包成Zip
        try (final ZipOutputStream zos = new ZipOutputStream(output)) {
            for (Path tempFile : tempFiles) {
                String fileName = tempFile.getFileName().toString();
                ZipEntry entry = new ZipEntry(fileName);
                zos.putNextEntry(entry);
                Files.copy(tempFile, zos);
                zos.closeEntry();
            }
        }
    } finally {
        // 清理临时文件
        for (Path tempFile : tempFiles) {
            Files.deleteIfExists(tempFile);
        }
    }
}

这种方式能确保每个数据源连接被最快释放，彻底避免超时问题，但需要注意磁盘空间是否足够。

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