java - 使用 Java future 的线程安全集合收集操作

Question

我第一次尝试使用 java future 。我设置了一个类来解压缩一些文件。想要递归解压缩，因为我有包含 zip 文件的 zip 文件。

我有一个驱动器类，它实例化一个实现可调用的 Uncompressor 类。 Uncompressor 开始解压缩，如果它遇到另一个 .zip，它会创建一个自己的实例，将它添加到池中，然后继续。

伪代码

From DriverClass:

.
.
.
ExecutorService pool = new Executors.newFixedThreadPool(4);
Uncompressor uc = new Uncompressor(pool, compressedFile);
Collection<File> files = uc.uncompress();
for(Future <Collection<File>> f : uc.futures)
    files.addAll(f.get());
// at the end of this loop, files doesnt seem to hold all of my files

这是我的解压器类

public class Uncompressor implements Callable<Collection<File>>
{
    public Set<Future<Collection<File>>> futures = new HashSet<Future<Collection<File>>>();
    File compressedFile;
public Uncompressor(ExecutorService pool, File compressedFile)
{
    this.pool = pool;
    this.compressedFile = compressedFile;
}

public Collection<File> call() throws Exception[
   return uncompress();
}

public Collection<File> uncompress()
{
List<File> uncompressedFiles = new ArrayList<File>();
.
.Loop
.//Try to uncompress the file. If the archive entry is a zip file, do the following:

    Callable<Collection<File>> callable = new Uncompressor(this.pool, archiveFileEntry);
    Future f = pool.submit(callable);
    futures.add(f);

 //else, add files to a collection here for returning
 uncompressedFiles.add(archiveFileEntry);

.EndLoop

return uncompressedFiles;
.
.
}

所以问题出在我的 DriverClass 中，我的文件集合应该包含来自递归潜水的所有未压缩文件，但似乎没有包含所有文件。我想我在从 Future 获取返回值时做错了。是不是因为我有一个类成员变量 futures的方式定义？

谢谢

Answer 1

如果嵌套深度大于 1，您的代码将无法工作，即顶级 zip 包含一个包含 zip 的 zip。在这种情况下，您的顶级解压缩器将没有底层 zip 文件的 future 。

使用 ForkJoinPool 会很酷为此，因为它更适合递归类型的任务分解。 Java 7 不是一个选项，我要做的是更改解压缩器，以便结果是 future 和文件的元组，然后更改调用者以跟踪所有未完成的 future :

--Caller--

Collection<File> alluncompressedFiles = new HashSet<File>();
Collection<Future<UncompressorResult>> futures = new LinkedList<Future<UncompressorResult>>();
Future<UncompressorResult> future = pool.submit(new Uncompressor(pool, compressedFile));
futures.add(future);

while (!futures.isEmpty()) {
    Future<UncompressorResult> future = futures.poll();
    UncompressorResult result = future.get();

    futures.addAll(result.getFutures());
    uncompressedFiles.addAll(result.getFiles());
}

并且 Uncompressor 更改为以下内容:

public UncompressorResult call() throws Exception[
    List<File> uncompressedFiles = new ArrayList<File>();

    for (File entry : zipFiles) {
        if (entry is not ZIP) {
            uncompressedFiles.add(entry);
        } else {
            Callable<UncompressorResult> callable = new Uncompressor(this.pool, entry);
            Future<UncompressorResult> f = pool.submit(callable);
            futures.add(f);
        }
    }

    return new UncompressorResult(uncompressedFiles, futures;
}