使用 CountDownLatch 时如何正确同步/锁定

Question

归结为一个线程通过某个服务提交作业。 Job 在一些 TPExecutor 中执行。之后，此服务在某些条件下（作业超过最大重试次数等）检查结果并在原始线程中抛出异常。下面的代码 sn-p 大致说明了遗留代码中的这种情况：

import java.util.concurrent.CountDownLatch;

public class IncorrectLockingExample {

private static class Request {

    private final CountDownLatch latch = new CountDownLatch(1);

    private Throwable throwable;

    public void await() {
        try {
            latch.await();
        } catch (InterruptedException ignoredForDemoPurposes) {
        }
    }

    public void countDown() {
        latch.countDown();
    }

    public Throwable getThrowable() {
        return throwable;
    }

    public void setThrowable(Throwable throwable) {
        this.throwable = throwable;
    }

}

private static final Request wrapper = new Request();

public static void main(String[] args) throws InterruptedException {

    final Thread blockedThread = new Thread() {
        public void run() {
            wrapper.await();
            synchronized (wrapper) {
                if (wrapper.getThrowable() != null)
                    throw new RuntimeException(wrapper.getThrowable());
            }
        }
    };

    final Thread workingThread = new Thread() {
        public void run() {
            wrapper.setThrowable(new RuntimeException());
            wrapper.countDown();

        }
    };

    blockedThread.start();
    workingThread.start();

    blockedThread.join();
    workingThread.join();
}

}

有时，（在我的机器上无法重现，但在 16 核服务器机器上发生）异常不会被报告给原始线程。我认为这是因为没有强制发生之前发生（例如，'countDown' 发生在 'setThrowable' 之前）并且程序继续工作（但应该失败）。 我将不胜感激有关如何解决此案的任何帮助。 限制条件是：一周内发布，需要对现有代码库的影响最小。

Answer 1

上面的代码（现在已更新）应该可以按您的预期工作，而无需使用进一步的同步机制。内存屏障及其对应的 'happens-before' 关系通过使用 CountDownLatch await() 和 countdown() 方法强制执行。

来自API docs：

“释放”同步器方法（例如 Lock.unlock、Semaphore.release 和 CountDownLatch.countDown）之前的操作发生在成功“获取”方法（例如 Lock.lock、Semaphore.acquire）之后的操作之前， Condition.await 和 CountDownLatch.await 在另一个线程中的同一个同步器对象上。

如果您经常处理并发问题，请为自己准备一份 'Java Concurrency in Practice'，它是 Java 并发圣经，非常值得放在您的书架上 :-)。

Answer 2

我怀疑你需要

private volatile Throwable throwable

您是否尝试过使用内置的 ExecutorService 并为您执行此操作。以下打印件

future1 := result
future2  threw java.lang.IllegalStateException
future3  timed out

代码是

public static void main(String... args)  {
    ExecutorService executor = Executors.newSingleThreadExecutor();
    Future<String> future1 = executor.submit(new Callable<String>() {
        public String call() throws Exception {
            return "result";
        }
    });

    Future<String> future2 = executor.submit(new Callable<String>() {
        public String call() throws Exception {
            throw new IllegalStateException();
        }
    });

    Future<String> future3 = executor.submit(new Callable<String>() {
        public String call() throws Exception {
            Thread.sleep(2000);
            throw new AssertionError();
        }
    });

    printResult("future1", future1);
    printResult("future2", future2);
    printResult("future3", future3);
    executor.shutdown();
}

private static void printResult(String description, Future<String> future) {
    try {
        System.out.println(description+" := "+future.get(1, TimeUnit.SECONDS));
    } catch (InterruptedException e) {
        System.out.println(description+"  interrupted");
    } catch (ExecutionException e) {
        System.out.println(description+"  threw "+e.getCause());
    } catch (TimeoutException e) {
        System.out.println(description+"  timed out");
    }
}

在 FutureTask 的代码中，有一个注释。

/**
 * The thread running task. When nulled after set/cancel, this
 * indicates that the results are accessible.  Must be
 * volatile, to ensure visibility upon completion.
 */

如果您不打算重用 JDK 中的代码，它仍然值得一读，以便您了解他们使用的任何技巧。