如果还没有进行则开始执行，否则等待现有执行在并发 Java 方法调用中完成

Question

假设我有一个 class 可以同时调用其方法 (process)。这个方法做了一些不能并发执行的处理（doProcess）。此外，如果调用了 process 方法，但处理 (doProcess) 已经在进行中 - process 的调用应该等到已经开始的处理 (doProcess) 完成并且然后 return（不开始新的处理 - doProcess）。如果 none 个线程正在执行实际处理 (doProcess)，则调用 process 方法只会执行实际处理 (doProcess)。

// Just imaginary example to illustrate the idea
class MyResource() {
   val lock = MyLock()
   fun process() {
      if (!lock.isLocked) {
         lock.lock()
         doProcess()
         lock.unlock()
      }
      lock.awaitUnlocked()
   }
   private fun doProcess() {
      // actual processing
   }
}

上面假想示例中描述的并发问题的预期并发 (java.util.concurrent) 原语是什么？我想 ReentrantLock 但它不提供 awaitUnlocked。对于这个简单的用例，java.util.concurrent.locks.Condition 似乎太低级了。

Answer 1

听起来您所追求的可以通过 ConcurrentMap 来实现——更具体地说，是 computeIfAbsent。

例如，您可以运行下面的代码。有点像缓存 - 如果你想要的东西可用，只需 return 它。否则等待哪个线程先到达那里先计算它。

import java.time.Instant;
import java.util.concurrent.*;

public class ConcurrencyTest
{
    private final ConcurrentMap<String, String> processItems = new ConcurrentHashMap<>();

    public String process()
    {
        return processItems.computeIfAbsent("doProcess", k -> doProcess());
    }

    private String doProcess()
    {
        try
        {
            System.out.println(Instant.now() + " -> doProcess() Starting some work on " + Thread.currentThread());
            Thread.sleep(100);
            System.out.println(Instant.now() + " -> doProcess() Finished some work on " + Thread.currentThread());
            return Instant.now().toString();
        }
        catch(Exception e)
        {
            throw new RuntimeException("Unexpected Exception : " + e, e);
        }
    }

    /**
     * And to test it out.
     *
     * @param args
     * @throws Exception
     */
    public static void main(String... args) throws Exception
    {
        final int THREADS = 10;
        final ConcurrencyTest test = new ConcurrencyTest();
        final ExecutorService execs = Executors.newFixedThreadPool(THREADS);
        final CountDownLatch startingGun = new CountDownLatch(1);

        for (int i = 0; i < THREADS; i++)
        {
            execs.submit(() -> {
                System.out.println(Instant.now() + " -> Thread -> " + Thread.currentThread() + " - Awaiting Starting Gun");
                try
                {
                    startingGun.await();
                }
                catch (InterruptedException e)
                {
                    throw new RuntimeException("Failure waiting for the starting gun.");
                }
                System.out.println(Instant.now() + " -> Running Thread -> " + Thread.currentThread());
                String val = test.process();
                System.out.println(Instant.now() + " -> Got back " + val + " -> " + Thread.currentThread());
            });
        }

        System.out.println("All tasks submitted.. waiting for 5 seconds then firing the starting gun. ");
        Thread.sleep(5_000);
        startingGun.countDown();
        execs.shutdown();
    }

}

获取以下输出。如您所见，只有一个线程最终执行了相关代码。剩下的等着吧。

All tasks submitted.. waiting for 5 seconds then firing the starting gun. 
2021-12-30T17:59:51.696626Z -> Thread -> Thread[pool-1-thread-3,5,main] - Awaiting Starting Gun
2021-12-30T17:59:51.696606Z -> Thread -> Thread[pool-1-thread-8,5,main] - Awaiting Starting Gun
2021-12-30T17:59:51.696245Z -> Thread -> Thread[pool-1-thread-10,5,main] - Awaiting Starting Gun
2021-12-30T17:59:51.696231Z -> Thread -> Thread[pool-1-thread-6,5,main] - Awaiting Starting Gun
2021-12-30T17:59:51.696627Z -> Thread -> Thread[pool-1-thread-5,5,main] - Awaiting Starting Gun
2021-12-30T17:59:51.696633Z -> Thread -> Thread[pool-1-thread-4,5,main] - Awaiting Starting Gun
2021-12-30T17:59:51.696274Z -> Thread -> Thread[pool-1-thread-2,5,main] - Awaiting Starting Gun
2021-12-30T17:59:51.696231Z -> Thread -> Thread[pool-1-thread-1,5,main] - Awaiting Starting Gun
2021-12-30T17:59:51.696620Z -> Thread -> Thread[pool-1-thread-7,5,main] - Awaiting Starting Gun
2021-12-30T17:59:51.696261Z -> Thread -> Thread[pool-1-thread-9,5,main] - Awaiting Starting Gun
2021-12-30T17:59:56.695927Z -> Running Thread -> Thread[pool-1-thread-5,5,main]
2021-12-30T17:59:56.696031Z -> Running Thread -> Thread[pool-1-thread-6,5,main]
2021-12-30T17:59:56.695968Z -> Running Thread -> Thread[pool-1-thread-4,5,main]
2021-12-30T17:59:56.695863Z -> Running Thread -> Thread[pool-1-thread-10,5,main]
2021-12-30T17:59:56.695832Z -> Running Thread -> Thread[pool-1-thread-8,5,main]
2021-12-30T17:59:56.696063Z -> Running Thread -> Thread[pool-1-thread-2,5,main]
2021-12-30T17:59:56.696133Z -> Running Thread -> Thread[pool-1-thread-3,5,main]
2021-12-30T17:59:56.696254Z -> Running Thread -> Thread[pool-1-thread-9,5,main]
2021-12-30T17:59:56.696230Z -> Running Thread -> Thread[pool-1-thread-7,5,main]
2021-12-30T17:59:56.696204Z -> Running Thread -> Thread[pool-1-thread-1,5,main]
2021-12-30T17:59:56.714154Z -> doProcess() Starting some work on Thread[pool-1-thread-4,5,main]
2021-12-30T17:59:56.814608Z -> doProcess() Finished some work on Thread[pool-1-thread-4,5,main]
2021-12-30T17:59:56.815375Z -> Got back 2021-12-30T17:59:56.815016Z -> Thread[pool-1-thread-5,5,main]
2021-12-30T17:59:56.815422Z -> Got back 2021-12-30T17:59:56.815016Z -> Thread[pool-1-thread-7,5,main]
2021-12-30T17:59:56.815104Z -> Got back 2021-12-30T17:59:56.815016Z -> Thread[pool-1-thread-8,5,main]
2021-12-30T17:59:56.815065Z -> Got back 2021-12-30T17:59:56.815016Z -> Thread[pool-1-thread-3,5,main]
2021-12-30T17:59:56.815093Z -> Got back 2021-12-30T17:59:56.815016Z -> Thread[pool-1-thread-2,5,main]
2021-12-30T17:59:56.815054Z -> Got back 2021-12-30T17:59:56.815016Z -> Thread[pool-1-thread-4,5,main]
2021-12-30T17:59:56.815420Z -> Got back 2021-12-30T17:59:56.815016Z -> Thread[pool-1-thread-6,5,main]
2021-12-30T17:59:56.815387Z -> Got back 2021-12-30T17:59:56.815016Z -> Thread[pool-1-thread-1,5,main]
2021-12-30T17:59:56.815077Z -> Got back 2021-12-30T17:59:56.815016Z -> Thread[pool-1-thread-9,5,main]
2021-12-30T17:59:56.815435Z -> Got back 2021-12-30T17:59:56.815016Z -> Thread[pool-1-thread-10,5,main]

如果还没有进行则开始执行，否则等待现有执行在并发 Java 方法调用中完成

Start execution if it is not in progress yet, otherwise wait for the existing execution to finish in concurrent Java method call

java

concurrency

synchronization

java.util.concurrent

kotlin