无锁解决方案（？）
我遇到了同样的问题，但我想要一个不使用锁的解决方案。
问题：我最多有一个线程从队列中消耗。多个生产者线程不断地插入队列，如果消费者在等待，则需要通知消费者。队列是无锁的，因此使用锁进行通知会导致生产者线程中不必要的阻塞。每个生产者线程都需要获取锁才能通知等待的消费者。我相信我想出了一个使用LockSupport 和AtomicReferenceFieldUpdater 的无锁解决方案。如果 JDK 中存在无锁屏障，我找不到它。 CyclicBarrier 和 CoundDownLatch 在内部都使用我能找到的锁。
这是我稍微缩写的代码。为了清楚起见，这段代码一次只允许 一个 线程等待。通过使用某种类型的原子集合来存储多个所有者，可以对其进行修改以允许多个等待者/消费者（ConcurrentMap 可能有效）。
我已经使用了这段代码，它似乎可以工作。我没有对它进行广泛的测试。我建议您在使用前阅读LockSupport 的文档。
/* I release this code into the public domain.
* http://unlicense.org/UNLICENSE
*/
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.LockSupport;
/**
* A simple barrier for awaiting a signal.
* Only one thread at a time may await the signal.
*/
public class SignalBarrier {
/**
* The Thread that is currently awaiting the signal.
* !!! Don't call this directly !!!
*/
@SuppressWarnings("unused")
private volatile Thread _owner;
/** Used to update the owner atomically */
private static final AtomicReferenceFieldUpdater<SignalBarrier, Thread> ownerAccess =
AtomicReferenceFieldUpdater.newUpdater(SignalBarrier.class, Thread.class, "_owner");
/** Create a new SignalBarrier without an owner. */
public SignalBarrier() {
_owner = null;
}
/**
* Signal the owner that the barrier is ready.
* This has no effect if the SignalBarrer is unowned.
*/
public void signal() {
// Remove the current owner of this barrier.
Thread t = ownerAccess.getAndSet(this, null);
// If the owner wasn't null, unpark it.
if (t != null) {
LockSupport.unpark(t);
}
}
/**
* Claim the SignalBarrier and block until signaled.
*
* @throws IllegalStateException If the SignalBarrier already has an owner.
* @throws InterruptedException If the thread is interrupted while waiting.
*/
public void await() throws InterruptedException {
// Get the thread that would like to await the signal.
Thread t = Thread.currentThread();
// If a thread is attempting to await, the current owner should be null.
if (!ownerAccess.compareAndSet(this, null, t)) {
throw new IllegalStateException("A second thread tried to acquire a signal barrier that is already owned.");
}
// The current thread has taken ownership of this barrier.
// Park the current thread until the signal. Record this
// signal barrier as the 'blocker'.
LockSupport.park(this);
// If a thread has called #signal() the owner should already be null.
// However the documentation for LockSupport.unpark makes it clear that
// threads can wake up for absolutely no reason. Do a compare and set
// to make sure we don't wipe out a new owner, keeping in mind that only
// thread should be awaiting at any given moment!
ownerAccess.compareAndSet(this, t, null);
// Check to see if we've been unparked because of a thread interrupt.
if (t.isInterrupted())
throw new InterruptedException();
}
/**
* Claim the SignalBarrier and block until signaled or the timeout expires.
*
* @throws IllegalStateException If the SignalBarrier already has an owner.
* @throws InterruptedException If the thread is interrupted while waiting.
*
* @param timeout The timeout duration in nanoseconds.
* @return The timeout minus the number of nanoseconds that passed while waiting.
*/
public long awaitNanos(long timeout) throws InterruptedException {
if (timeout <= 0)
return 0;
// Get the thread that would like to await the signal.
Thread t = Thread.currentThread();
// If a thread is attempting to await, the current owner should be null.
if (!ownerAccess.compareAndSet(this, null, t)) {
throw new IllegalStateException("A second thread tried to acquire a signal barrier is already owned.");
}
// The current thread owns this barrier.
// Park the current thread until the signal. Record this
// signal barrier as the 'blocker'.
// Time the park.
long start = System.nanoTime();
LockSupport.parkNanos(this, timeout);
ownerAccess.compareAndSet(this, t, null);
long stop = System.nanoTime();
// Check to see if we've been unparked because of a thread interrupt.
if (t.isInterrupted())
throw new InterruptedException();
// Return the number of nanoseconds left in the timeout after what we
// just waited.
return Math.max(timeout - stop + start, 0L);
}
}
为了给出一个模糊的用法示例，我将采用 james large 的示例：
SignalBarrier barrier = new SignalBarrier();
消费线程（单数，不是复数！）：
try {
while(!conditionIsTrue()) {
barrier.await();
}
doSomethingThatRequiresConditionToBeTrue();
} catch (InterruptedException e) {
handleInterruption();
}
生产者线程：
doSomethingThatMakesConditionTrue();
barrier.signal();