如何在没有队列的情况下实现这个并发结构？答案

【问题标题】：How can I implement this concurrent structure without the queue?如何在没有队列的情况下实现这个并发结构？
【发布时间】：2019-11-13 02:57:10
【问题描述】：

我的应用程序中有一种情况，即事件进入并且处理它们的线程（信号线程）必须向另一个线程（工作线程）发出信号，到目前为止它处于空闲状态，它可以运行一些代码。一旦工作线程完成，它应该等待再次发出信号。工作线程正在工作时，事件可能会到达。在这种情况下，它应该继续前进并立即继续工作。工作线程的一个动作对任何数量的传入事件都足够工作，因此不需要每个事件工作一次，只需在每个事件之后尽快工作一次。正确行为示例：

event comes in
worker thread starts work
worker thread finishes work
event comes in
worker thread starts work
event comes in
event comes in
worker thread finishes work
worker thread starts work
worker thread finishes work

4 个事件，3 个工作周期。这是一个不幸但不可避免的要求，即信号线程在处理事件时不能阻塞。目前我已经使用 BlockingQueue 实现了这一点，它具有填充自身的无意义的副作用，即使内容不有趣甚至看起来也不有趣。我希望能够使用 CountDownLatch 或 CyclicBarrier 或类似方法来完成这项工作，但我一直找不到方法。这是我的实现：

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class Main {

    private static final class MyBarrier {
        private BlockingQueue<Boolean> queue = new LinkedBlockingQueue<>();
        void await() throws InterruptedException {
            queue.take();
            queue.clear();
        }
        void signal() {
            queue.add(true);
        }
    }

    private static Random random = new Random(0);

    private static void sleepForMax(int maxMillis) {
        sleep(random.nextInt(maxMillis));
    }

    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        MyBarrier myBarrier = new MyBarrier();
        final ExecutorService singallingThread = Executors.newSingleThreadExecutor();
        singallingThread.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                sleepForMax(1_000); // simulate period between events arriving
                myBarrier.signal();
                System.out.println("Signalling work to be done");
            }
            System.out.println("Thread interrupted");
        });
        final ExecutorService workingThread = Executors.newSingleThreadExecutor();
        workingThread.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    System.out.println("Waiting for work");
                    myBarrier.await();
                } catch (InterruptedException e) {
                    break;
                }
                System.out.println("Doing work...");
                sleepForMax(3_000); // simulate work being done
                System.out.println("Work done");
            }
            System.out.println("Thread interrupted");
        });
        sleep(10_000);
        singallingThread.shutdownNow();
        workingThread.shutdownNow();
    }

}

有什么更好的方法来做到这一点？

【问题讨论】：

标签： java multithreading concurrency java.util.concurrent

【解决方案1】：

当我使用您使用 Phaser 的实现运行您的代码时，已经更改了睡眠时间，以便每 800 毫秒发生一次信号发送并且处理需要 1000 毫秒，例如，我得到了这个输出：

00008: Waiting for work
00808: Signalling work to be done
00808: Doing work...                      <-- worker starts working
01608: Signalling work to be done         <-- signal came, so there's more work
01808: Work done
01809: Waiting for work                   <-- waits for work...
02409: Signalling work to be done         <-- ...for 600 ms, until the next signal
02409: Doing work...

（左边的数字是自开始以来的毫秒数。此外，您可以使用随机延迟的代码重现它，但这更难重现和看到。）

如果我理解正确，那就错了。例如。想象一下如果信号停止发出会发生什么。

您的代码可能可以针对您的特定情况进行此调整：

private static final class MyBarrierWithPhaser {

    private final Phaser phaser = new Phaser(1);
    private int lastObservedPhase; // Phaser has initial phase 0

    void await() throws InterruptedException {
        // only works for 1 producer 1 worker; lastObservedPhase is kind of thread-local
        lastObservedPhase = phaser.awaitAdvanceInterruptibly(lastObservedPhase);
    }

    void signal() {
        phaser.arrive();
    }
}

这样，worker 记录了它前进到的最后一个阶段，如果信号线程在下一个awaitAdvanceInterruptibly 之前“到达”，那么 Phaser 阶段会被更新，并且当 worker 尝试使用过时的阶段等待时，它将立即取得进展；如果信号线程在awaitAdvanceInterruptibly之前没有到达，那么worker会一直等到信号线程最终到达。

使用更简单的同步原语，我可以想到如何使用synchronized-wait()-notify()机制来实现：

private static final class MyBarrierWithSynchronized {

    private boolean hasWork = false;

    synchronized void await() throws InterruptedException {
        while (!hasWork) {
            wait();
        }
        hasWork = false;
    }

    synchronized void signal() {
        hasWork = true;
        notifyAll(); // or notify() if we are sure there is 1 signal thread and 1 worker thread
    }
}

它有几个缺点：await() 在线程等待进入时不会被中断。另外，有些人不喜欢在this 上同步，为了简短起见，我保留了它。这可以使用java.util.concurrent.* 类似物重写，这种实现不会同时具有这两个缺点：

private static final class MyBarrierWithLock {

    private boolean hasWorkFlag = false;

    private final Lock lock = new ReentrantLock();
    private final Condition hasWorkCond = lock.newCondition();

    void await() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (!hasWorkFlag) {
                hasWorkCond.await();
            }
            hasWorkFlag = false;
        } finally {
            lock.unlock();
        }
    }

    void signal() {
        lock.lock();
        try {
            hasWorkFlag = true;
            hasWorkCond.signalAll(); // or signal() if we are sure there is 1 signal thread and 1 worker thread
        } finally {
            lock.unlock();
        }
    }
}

【讨论】：

谢谢！我可以确切地看到你的意思的错误。我将在您的更新中使用 Phaser 解决方案。在我的实际代码中，它更适合（不需要单独的类）。谢谢！

【解决方案2】：

我正在对此进行试验，使用 java.util.concurrent.Phaser，它可能有效，但我之前没有使用过 Phaser，所以我不确定。

private static final class MyBarrier2 {
    private Phaser phaser = new Phaser(1);
    void await() throws InterruptedException {
        phaser.awaitAdvanceInterruptibly(phaser.getPhase());
    }
    void signal() {
        phaser.arrive();
    }
}

【讨论】：