執行緒通訊用來確保執行緒協調運行,一般在做執行緒同步的時候才需要考慮執行緒通訊的問題。
1、傳統的執行緒通訊
通常利用Objeclt類別提供的三個方法:
wait() 導致當前執行緒等待,並釋放該同步監視器的鎖定,直到其它執行緒呼叫該同步監視器的notify( )或者notifyAll()方法喚醒執行緒。
notify(),喚醒在此同步監視器上等待的線程,如果有多個會任意選擇一個喚醒
notifyAll() 喚醒在此同步監視器上等待的所有線程,這些線程通過調度競爭資源後,某個執行緒取得此同步監視器的鎖,然後得以運作。
這三個方法必須由同步監視器物件調用,分為兩張情況:
同步方法時,由於同步監視器為this對象,所以可以直接調用這三個方法。
範例如下:
public class SyncMethodThreadCommunication { static class DataWrap{ int data = 0; boolean flag = false; public synchronized void addThreadA(){ if (flag) { try { wait(); } catch (InterruptedException e) { e.printStackTrace(); } } data++; System.out.println(Thread.currentThread().getName() + " " + data); flag = true; notify(); } public synchronized void addThreadB() { if (!flag) { try { wait(); } catch (InterruptedException e) { e.printStackTrace(); } } data++; System.out.println(Thread.currentThread().getName() + " " + data); flag = false; notify(); } } static class ThreadA extends Thread { private DataWrap data; public ThreadA(DataWrap dataWrap) { this.data = dataWrap; } @Override public void run() { for (int i = 0; i < 10; i++) { data.addThreadA(); } } } static class ThreadB extends Thread { private DataWrap data; public ThreadB(DataWrap dataWrap) { this.data = dataWrap; } @Override public void run() { for (int i = 0; i < 10; i++) { data.addThreadB(); } } } public static void main(String[] args) { //实现两个线程轮流对数据进行加一操作 DataWrap dataWrap = new DataWrap(); new ThreadA(dataWrap).start(); new ThreadB(dataWrap).start(); } }
同步程式碼區塊時,需要使用監視器物件呼叫這三個方法。
範例如下:
public class SyncBlockThreadComminication { static class DataWrap{ boolean flag; int data; } static class ThreadA extends Thread{ DataWrap dataWrap; public ThreadA(DataWrap dataWrap){ this.dataWrap = dataWrap; } @Override public void run() { for(int i = 0 ; i < 10; i++) { synchronized (dataWrap) { if (dataWrap.flag) { try { dataWrap.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } dataWrap.data++; System.out.println(getName() + " " + dataWrap.data); dataWrap.flag = true; dataWrap.notify(); } } } } static class ThreadB extends Thread{ DataWrap dataWrap; public ThreadB(DataWrap dataWrap){ this.dataWrap = dataWrap; } @Override public void run() { for (int i = 0; i < 10; i++) { synchronized (dataWrap) { if (!dataWrap.flag) { try { dataWrap.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } dataWrap.data++; System.out.println(getName() + " " + dataWrap.data); dataWrap.flag = false; dataWrap.notify(); } } } } public static void main(String[] args) { //实现两个线程轮流对数据进行加一操作 DataWrap dataWrap = new DataWrap(); new ThreadA(dataWrap).start(); new ThreadB(dataWrap).start(); } }
2、使用Condition控制執行緒通訊
當使用Lock物件保證同步時,則使用Condition物件來確保協調。
範例如下:
import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import com.sun.media.sound.RIFFInvalidDataException; import javafx.scene.chart.PieChart.Data; public class SyncLockThreadCommunication { static class DataWrap { int data; boolean flag; private final Lock lock = new ReentrantLock(); private final Condition condition = lock.newCondition(); public void addThreadA() { lock.lock(); try { if (flag) { try { condition.await(); } catch (InterruptedException e) { e.printStackTrace(); } } data++; System.out.println(Thread.currentThread().getName() + " " + data); flag = true; condition.signal(); } finally { lock.unlock(); } } public void addThreadB() { lock.lock(); try { if (!flag) { try { condition.await(); } catch (InterruptedException e) { e.printStackTrace(); } } data++; System.out.println(Thread.currentThread().getName() + " " + data); flag = false; condition.signal(); } finally { lock.unlock(); } } } static class ThreadA extends Thread{ DataWrap dataWrap; public ThreadA(DataWrap dataWrap) { this.dataWrap = dataWrap; } @Override public void run() { for (int i = 0; i < 10; i++) { dataWrap.addThreadA(); } } } static class ThreadB extends Thread{ DataWrap dataWrap; public ThreadB(DataWrap dataWrap) { this.dataWrap = dataWrap; } @Override public void run() { for (int i = 0; i < 10; i++) { dataWrap.addThreadB(); } } } public static void main(String[] args) { //实现两个线程轮流对数据进行加一操作 DataWrap dataWrap = new DataWrap(); new ThreadA(dataWrap).start(); new ThreadB(dataWrap).start(); } }
其中Condition物件的await(), singal(),singalAll()分別對應wait(),notify()和notifyAll()方法。
3、使用阻塞隊列BlockingQueue控制線程通信
BlockingQueue是Queue接口的子接口,主要用來做線程通信使用,它具有一個特徵:當生產者線程試圖向BlockingQueue中放入元素時,如果隊列已放入元素時,如果隊列已放入滿,則該線程被阻塞;當消費者線程試圖從BlockingQueue中取出元素時,如果隊列已空,則該線程被阻塞。這兩個特徵分別對應兩個支援阻塞的方法,put(E e)和take()
範例如下:
import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; public class BlockingQueueThreadComminication { static class DataWrap{ int data; } static class ThreadA extends Thread{ private BlockingQueue<DataWrap> blockingQueue; public ThreadA(BlockingQueue<DataWrap> blockingQueue, String name) { super(name); this.blockingQueue = blockingQueue; } @Override public void run() { for (int i = 0; i < 100; i++) { try { DataWrap dataWrap = blockingQueue.take(); dataWrap.data++; System.out.println(getName() + " " + dataWrap.data); sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } } static class ThreadB extends Thread{ private BlockingQueue<DataWrap> blockingQueue; private DataWrap dataWrap; public ThreadB(BlockingQueue<DataWrap> blockingQueue, DataWrap dataWrap, String name) { super(name); this.blockingQueue = blockingQueue; this.dataWrap = dataWrap; } @Override public void run() { for (int i = 0; i < 100; i++) { try { dataWrap.data++; System.out.println(getName() + " " + dataWrap.data); blockingQueue.put(dataWrap); sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } } public static void main(String[] args) { ///实现两个线程轮流对数据进行加一操作 DataWrap dataWrap = new DataWrap(); BlockingQueue<DataWrap> blockingQueue = new ArrayBlockingQueue<>(1); new ThreadA(blockingQueue, "Consumer").start(); new ThreadB(blockingQueue, dataWrap, "Producer").start(); } }
BlockingQueue共有五個實作類別:
ArrayBlockingQueue 基於陣列為基礎實作的BlockingQueueue
ArrayBlockingQueue 基於陣列鍊錶實作的BlockingQueue佇列PriorityBlockingQueue 中元素需實作Comparable接口,其中元素的排序是依照Comparator進行的客製化排序。 SynchronousQueue 同步佇列,要求對該佇列的存取操作必須交替進行。 DelayQueue 集合元素必須實作Delay接口,佇列中元素排序依照Delay介面方法getDelay()的回傳值排序。