Iterate thru a HashMap and start a Thread for every Different Value

I need to write code that processes a HashMap (with two different classes being its key and value) and creates a thread for each of its items. To execute a thread, the code will take an entry from that HashMap and check if there is a thread with the same value. If it exists, it will skip this entry until that thread finishes executing. The code must go through this HashMap until it is empty, without breaking the condition I mentioned above. However, I am having trouble implementing this logic.

I created an Executor Service based on the number of processors and a list of values (from HashMap) that are being executed. But I don't know how I can remove the value from this list after the thread has finished running.

Here's my unfinished and untested code:

    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();
    }

I think you have a bit of an XY problem there. It looks to me as if your actual requirement is to ensure that no two threads run ConfirmNfeProcessor tasks for the same dealership at the same time, is that right?

Instead of having one fixed thread pool with N worker threads, I would use an array of N thread "pools" that have one thread each. Then, for each pair in the map, I would use pair.getValue().hashCode() to choose which one of the N executors should perform the task.

Every pair that has the same value ("value" == "dealership", right?) will be submitted to the same executor, and since each executor is single-threaded, it will be guaranteed to never perform two tasks for the same dealership at the same time.

The simplest way to achieve this, is to use CompletableFuture, which allows you to chain dependent actions, like submitting another job, when the asynchronous evaluation has been completed.

E.g.

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();

It uses a Map<UnidadeOrganizacionalView, CompletableFuture<Boolean>> to remember the already submitted jobs.

This is used within the compute method. It will call CompletableFuture.supplyAsync(p, executor) when there was no previous job associated with the value, to create a new one and remember it. Otherwise, it will create a new job via thenApplyAsync, that will be started when the previous one has been completed, and remembers this new job.

This assumes that it is possible to change the class ConfirmNfeProcessor from implementing Callable<Boolean> to implementing Supplier<Boolean>. The main difference, besides the name of the relevant method, is that a Supplier can not throw checked exceptions. If such a change is not possible, you need adapter code.

One possibility would be:

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;
}

to be used like

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));
}