ZipOutputStream阻塞Vertx事件循环

import io.vertx.core.Vertx;
import io.vertx.core.file.FileSystem;
import io.vertx.core.file.OpenOptions;
import io.vertx.core.streams.Pump;
import io.vertx.core.streams.ReadStream;

import java.io.FileOutputStream;
import java.io.IOException;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

public class ZipExample {

    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        FileSystem fileSystem = vertx.fileSystem();

        // 创建一个Worker Executor
        WorkerExecutor executor = vertx.createSharedWorkerExecutor("zip-worker-pool");

        // 在Worker Executor中执行压缩操作
        executor.executeBlocking(future -> {
            try {
                // 创建ZipOutputStream
                ZipOutputStream zipOutputStream = new ZipOutputStream(new FileOutputStream("output.zip"));

                // 获取要压缩的文件列表
                ReadStream<io.vertx.core.file.FileProps> fileListStream = fileSystem.propsStream("path/to/files");

                // 使用Pump将文件流写入ZipOutputStream
                Pump pump = Pump.pump(fileListStream, new WriteStream(zipOutputStream));
                pump.start();

                // 处理文件流结束事件
                fileListStream.endHandler(end -> {
                    // 关闭ZipOutputStream
                    zipOutputStream.close();

                    // 完成Future
                    future.complete();
                });

                // 处理文件流中的每个文件
                fileListStream.handler(fileProps -> {
                    try {
                        // 创建ZipEntry
                        ZipEntry zipEntry = new ZipEntry(fileProps.path());

                        // 将ZipEntry写入ZipOutputStream
                        zipOutputStream.putNextEntry(zipEntry);

                        // 读取文件内容并写入ZipOutputStream
                        ReadStream<Buffer> fileStream = fileSystem.openReadStream(fileProps.path(), new OpenOptions());
                        Pump.pump(fileStream, new WriteStream(zipOutputStream)).start();
                        fileStream.endHandler(end -> zipOutputStream.closeEntry());
                    } catch (IOException e) {
                        future.fail(e);
                    }
                });
            } catch (IOException e) {
                future.fail(e);
            }
        }, result -> {
            if (result.succeeded()) {
                System.out.println("压缩完成");
            } else {
                System.err.println("压缩失败：" + result.cause().getMessage());
            }

            // 关闭Worker Executor
            executor.close();
            vertx.close();
        });
    }

    // 自定义WriteStream，用于将数据写入ZipOutputStream
    private static class WriteStream implements io.vertx.core.streams.WriteStream<Buffer> {
        private final ZipOutputStream zipOutputStream;

        public WriteStream(ZipOutputStream zipOutputStream) {
            this.zipOutputStream = zipOutputStream;
        }

        @Override
        public io.vertx.core.streams.WriteStream<Buffer> exceptionHandler(Handler<Throwable> handler) {
            return this;
        }

        @Override
        public io.vertx.core.streams.WriteStream<Buffer> write(Buffer data) {
            try {
                zipOutputStream.write(data.getBytes());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            return this;
        }

        @Override
        public void end() {
            // Do nothing
        }

        @Override
        public io.vertx.core.streams.WriteStream<Buffer> setWriteQueueMaxSize(int maxSize) {
            return this;
        }

        @Override
        public boolean writeQueueFull() {
            return false;
        }

        @Override
        public io.vertx.core.streams.WriteStream<Buffer> drainHandler(Handler<Void> handler) {
            return this;
        }
    }
}