遍历 HashMap 并为每个不同的值启动一个线程答案

【问题标题】：Iterate thru a HashMap and start a Thread for every Different Value遍历 HashMap 并为每个不同的值启动一个线程
【发布时间】：2020-08-31 20:47:04
【问题描述】：

我需要编写代码来处理一个 HashMap（两个不同的类是它的键和值）并为它的每个项目创建一个线程。要执行线程，代码将从该 HashMap 中获取一个条目并检查是否存在具有相同值的线程。如果存在，它将跳过该条目，直到该线程完成执行。代码必须经过这个HashMap，直到它为空，而不破坏我上面提到的条件。但是，我在实现这个逻辑时遇到了麻烦。

我根据处理器数量和正在执行的值列表（来自 HashMap）创建了一个执行器服务。但我不知道如何在线程完成运行后从该列表中删除该值。

这是我未完成和未经测试的代码：

    int threads = Runtime.getRuntime().availableProcessors();
    ExecutorService executor = Executors.newFixedThreadPool(threads);
    List<Future<Boolean>> futures = new ArrayList<>();
    List<UnidadeOrganizacionalView> dealershipsBeingExecuted = new ArrayList<>();
    dealershipsBeingExecuted = Collections.synchronizedList(dealershipsBeingExecuted);

    Iterator<Map.Entry<NotaResumoView, UnidadeOrganizacionalView>> it =
            preparedNfs.entrySet().iterator();
    while (it.hasNext()) {
        Map.Entry<NotaResumoView, UnidadeOrganizacionalView> pair = it.next();
        if (dealershipsBeingExecuted.contains(pair.getValue())){
            it.next();
        }
        dealershipsBeingExecuted.add(pair.getValue());
        futures.add(executor.submit(
                new ConfirmNfeProcessor(Integration, pair.getKey(), pair.getValue())););
        it.remove();
    }

【问题讨论】：

不要将您提交给ExecutorService的对象称为“线程”。这些对象是任务。 ExecutorService 创建自己的线程，用于执行您提交给它的任务。
您应该有一个Map<UnidadeOrganizacionalView,Future<Boolean>> pending，而不是List<Future<Boolean>> futures 和List<UnidadeOrganizacionalView> dealershipsBeingExecuted，它允许您获取已经存在的任务并根据需要等待其完成。
Solomon Slow，谢谢指正，我有这种调用任务线程的习惯，可能是我对 Java 8 多任务处理缺乏了解所致。
不相关：在我完成评论之前，您删除了另一个问题。问题是：将您的复杂业务代码分解为一个小示例正是您通常调试此类问题的方式，无论是否在这里询问它们。这就是本质：你从一个简约的东西开始，然后你慢慢地添加更多，直到你“遇到”你的错误。如果仅此一项并不能让您知道出了什么问题，那么您可以在这里分享一些东西。其他任何事情都是浪费时间。

标签： java multithreading concurrency hashmap

【解决方案1】：

实现此目的的最简单方法是使用CompletableFuture，它允许您在异步评估完成后链接相关操作，例如提交另一个作业。

例如

Map<UnidadeOrganizacionalView, CompletableFuture<Boolean>> pending = new HashMap<>();

for(Map.Entry<NotaResumoView, UnidadeOrganizacionalView> e: preparedNfs.entrySet()) {
    UnidadeOrganizacionalView value = e.getValue();
    ConfirmNfeProcessor p = new ConfirmNfeProcessor(Integration, e.getKey(), value);
    pending.compute(value, (key,future) -> future == null?
        CompletableFuture.supplyAsync(p, executor):
        future.thenApplyAsync(b -> p.get(), executor));
}

// if you want to wait for the completion of all job:
CompletableFuture.allOf(pending.values().toArray(new CompletableFuture<?>[0])).join();

它使用Map<UnidadeOrganizacionalView, CompletableFuture<Boolean>> 来记住已经提交的作业。

这在compute 方法中使用。当没有与该值关联的先前作业时，它将调用CompletableFuture.supplyAsync(p, executor)，以创建一个新作业并记住它。否则，它将通过thenApplyAsync 创建一个新作业，该作业将在前一个作业完成后启动，并记住这个新作业。

这假设可以将类ConfirmNfeProcessor 从实现Callable<Boolean> 更改为实现Supplier<Boolean>。除了相关方法的名称外，主要区别在于Supplier 不能抛出已检查的异常。如果无法进行此类更改，则需要适配器代码。

一种可能性是：

public static <R> CompletableFuture<R> callAsync(Callable<R> callable, Executor e) {
    CompletableFuture<R> cf = new CompletableFuture<>();
    CompletableFuture.runAsync(() -> {
        try { cf.complete(callable.call()); }
        catch(Throwable ex) { cf.completeExceptionally(ex); }
    }, e);
    return cf;
}
public static <R> CompletableFuture<R> thenCallAsync(
                  CompletableFuture<?> f, Callable<R> callable, Executor e) {
    CompletableFuture<R> cf = new CompletableFuture<>();
    f.whenCompleteAsync((value, t) -> {
        if(t != null) cf.completeExceptionally(t);
        else try { cf.complete(callable.call()); }
        catch(Throwable ex) { cf.completeExceptionally(ex); }
    }, e);
    return cf;
}

像这样使用

Map<UnidadeOrganizacionalView, CompletableFuture<Boolean>> pending = new HashMap<>();

for(Map.Entry<NotaResumoView, UnidadeOrganizacionalView> e: preparedNfs.entrySet()) {
    UnidadeOrganizacionalView value = e.getValue();
    ConfirmNfeProcessor p = new ConfirmNfeProcessor(Integration, e.getKey(), value);
    pending.compute(value, (key,future) -> future == null?
        callAsync(p, executor): thenCallAsync(future, p, executor));
}

【讨论】：

谢谢！我不记得 CompletableFuture 甚至没有考虑将它与价值相关联。这很有帮助。

【解决方案2】：

我认为你有一点XY problem。在我看来，您的实际要求是确保没有两个线程同时为同一经销商运行ConfirmNfeProcessor 任务，对吗？

我不会使用一个带有 N 个工作线程的固定线程池，而是使用一组 N 个线程“池”，每个线程“池”都有一个线程。然后，对于地图中的每个pair，我将使用pair.getValue().hashCode() 来选择N 个执行者中的哪一个应该执行任务。

每一个具有相同值的pair（"value" == "dealership"，对吗？）都会被提交给同一个执行器，并且由于每个执行器都是单线程的，因此可以保证永远不会执行两个同时为同一个经销商执行任务。

【讨论】：