Condition

來自專欄 Java之鏈

上篇文章提到一道網易面試題：假設有N個線程，依次列印0, 1, 2, N-1, N, N+1, N+2 ...

2N-1……1000。我給出了semaphore的方法，但我面試的時候是用Condition解決的（因為當時還對semaphore不熟：）。先看一段JDK文檔：

Conditions (also known as condition queues or condition variables) provide a means for one thread to suspend execution (to "wait") until notified by another thread that some state condition may now be true. Because access to this shared state information occurs in different threads, it must be protected, so a lock of some form is associated with the condition. The key property that waiting for a condition provides is that it atomically releases the associated lock and suspends the current thread, just like Object.wait.

A Condition instance is intrinsically bound to a lock. To obtain a Condition instance for a particular Lock instance use its newCondition() method.

Condition是線程級別的選擇語句，只有true或false兩種結果。如果不滿足條件則會suspend execution，暫停線程並且釋放鎖。condition是臨界資源的一種狀態，而臨界資源需要lock的保護，所以condition必須和lock綁定在一起，否則會拋出異常。

class Worker3 implements Runnable { private int x; // 初始值 private int co; // 線程數量，也是增長步調 private Lock lock; private Condition condition; public Worker3(int x, int co, Lock lock, Condition condition) { this.x = x; this.co = co; this.lock = lock; this.condition = condition; } @Override public void run() { lock.lock(); try { while (x < 1000) { while (x != ConditionTest.n + 1) { // 注釋1 condition.await(); } System.out.println(Thread.currentThread().getName() + " " + (++ConditionTest.n)); condition.signal(); // 注釋2 x += co; } } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } }}public volatile static int n = -1;public static void main(String[] args) { int co = ThreadLocalRandom.current().nextInt(10, 20); Lock lock = new ReentrantLock(); Condition condition = lock.newCondition(); for (int i = 0; i < co; i++) { Thread thread = new Thread(new Worker3(i, co, lock, condition)); thread.start(); }}

上面這段代碼看似正確，但運行失敗。現在分析失敗原因：假設有N個線程，其中N-1個處於await狀態，那麼第N個線程獲得鎖以後，能列印輸出，但運行到 condition.signal() 這一行，發出信號喚醒某一個線程，但被喚醒的線程可能不滿足條件變數，繼續await，這樣所有的線程都處於等待狀態，形成死鎖。

怎樣解決這個死鎖問題呢？有兩種方法：

1）在注釋1處，增加一行condition.signal()

2）把注釋2處的代碼更改為condition.signalAll()，喚醒所有線程，必然包括那個能改變條件變數的線程。

所以多個線程共享一個condition需要慎重，要確保能喚醒等待的線程（可能是滿足某個條件的等待線程）。

對比上篇文章中semaphore的實現，兩者最大不同是Condition是先判斷是否滿足條件，然後等待或者列印；而後者是先獲得Semaphore的許可，然後再判斷。

順便說下，在特殊情況下，線程數量是2，那麼上面一段代碼也能正確運行。

再來看一段官網上的代碼：

class BoundedBuffer { final Lock lock = new ReentrantLock(); final Condition notFull = lock.newCondition(); final Condition notEmpty = lock.newCondition(); final Object[] items = new Object[100]; int putptr, takeptr, count; public void put(Object x) throws InterruptedException { lock.lock(); try { while (count == items.length) notFull.await(); items[putptr] = x; if (++putptr == items.length) putptr = 0; ++count; notEmpty.signal(); } finally { lock.unlock(); } } public Object take() throws InterruptedException { lock.lock(); try { while (count == 0) notEmpty.await(); Object x = items[takeptr]; if (++takeptr == items.length) takeptr = 0; --count; notFull.signal(); return x; } finally { lock.unlock(); } } }

這段代碼有兩個Condition，與阻塞隊列極為相似，請比較閱讀ArrayBlockingQueue源碼。

上面代碼中兩個Condition變數（notFull ，notEmpty）的命名值得學習和品味。

其實還有一種樸素的方法也能實現

class Worker1 implements Runnable { private int x; private int co; private Lock lock; public Worker1(int x, int co, Lock lock) { this.x = x; this.co = co; this.lock = lock; } @Override public void run() { while (x < 1000) { if (x == ConditionTest.n + 1) { lock.lock(); System.out.println(Thread.currentThread().getName() +" " + (++ConditionTest.n)); lock.unlock(); x += co; } } }}public static void main(String[] args) throws InterruptedException { int co = ThreadLocalRandom.current().nextInt(10, 20); Lock lock = new ReentrantLock(); for (int i = 0; i < co; i++) { Thread thread = new Thread(new Worker1(i, co, lock)); thread.start(); }}

這種也能正確輸出，但是由於lock頻繁競爭，導致耗時明顯太長。

參考資料： Condition (Java Platform SE 8 )

推薦閱讀：