CompletableFuture将结果收集到hashmap

Question

我正在从数据库中读取pdf文件列表，对它们进行解析，并用它们执行一些任务
当我阅读这个pdf列表时，我发现从pdf中提取图像需要花费更多的时间，而且我不需要阻止我的主线程来阅读图像。所以我想在一个单独的线程中执行提取图像
我想从一个又一个pdf中读取图像，而不是一次将所有pdf加载到内存中（由于内存问题）。所以我只想要两条线；一个应该是主线程（从pdf中读取一些文本并执行其他操作），另一个应该是提取图像并返回图像对象集的线程。

这里的一个警告是，来自pdfs的图像在其内容中可能是相同的，所以我想在收集结果之前使用校验和或其他方法删除重复的图像。
我不想在内存中保存图像，直到所有任务完成，我当我得到一个pdf的结果时，想要删除重复的文件

所以真正的问题是，我需要向大小为1的线程池提交多个任务，并且需要在得到结果时删除重复项，这样我就不需要在内存中保存更长时间的映像。

我已经从代码中删除了不必要的东西，比如图像及其内容，并将代码转换为基于字符串的问题。

public static void main(String[] args) throws InterruptedException, ExecutionException {
        Map<String, Integer> uniqueImages = new HashMap<>();
         ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(1);
        List<CompletableFuture<String>> futureList = new ArrayList<>();
        for(int i = 0; i<20000; i++) {
            CompletableFuture<String> obj = CompletableFuture.supplyAsync(()->{
                //Assume lot of duplicates
                return UUID.randomUUID().toString();
            }, newFixedThreadPool).thenApply((x)->{
                if(uniqueImages.containsKey(x)) {
                    int val = uniqueImages.get(x);
                    uniqueImages.put(x, val+1);
                }
                else {
                    uniqueImages.put(x, 1);
                }
                return x;
            });
            futureList.add(obj);
        }
        
        for(CompletableFuture<String> future: futureList) {
            future.get();
        }
        System.out.println(uniqueImages.size());
    }

我担心这段代码是否真的有效或引发ConcurrentModification异常。

• 是否uniqueImagesmap真的包含唯一的图像及其计数？
• 有什么隐藏的问题吗？
• 有更好的方法来解决我的用例吗？

Answer 1

只要您只使用大小为1的线程池，就没有问题。

但是，如果有多个线程，那么代码确实不是线程安全的，并且容易出现数据竞争。
我们可以使用以下场景进行演示：

考虑以下代码：

if(uniqueImages.containsKey(x)) {
    int val = uniqueImages.get(x);
    uniqueImages.put(x, val+1);
}
else {
    uniqueImages.put(x, 1);
}

想象线程1和线程2都返回相同的字符串，并到达行if（uniqueImages.containsKey（x））在一起。
if将在两个线程中返回false，并且uniqueImages.put（x，1）；将在两个线程中被调用。
可能会抛出一个Con电流修改异常，或者您会得到一个错误的计数（1而不是2）。

如果计划使用多个线程，则必须使用ConcurrentHashMap。