Java并发编程(04):线程间通信,等待/通知机制
本文源碼:GitHub·點(diǎn)這里 || GitEE·點(diǎn)這里
一、概念簡介
1、線程通信
在操作系統(tǒng)中,線程是個獨(dú)立的個體,但是在線程執(zhí)行過程中,如果處理同一個業(yè)務(wù)邏輯,可能會產(chǎn)生資源爭搶,導(dǎo)致并發(fā)問題,通常使用互斥鎖來控制該邏輯。但是在還有這樣一類場景,任務(wù)執(zhí)行是有順序控制的,例如常見的報表數(shù)據(jù)生成:
- 啟動數(shù)據(jù)分析任務(wù),生成報表數(shù)據(jù);
- 報表數(shù)據(jù)存入指定位置數(shù)據(jù)容器;
- 通知數(shù)據(jù)搬運(yùn)任務(wù),把數(shù)據(jù)寫入報表庫;
該場景在相對復(fù)雜的系統(tǒng)中非常常見,如果基于多線程來描述該過程,則需要線程之間通信協(xié)作,才能有條不紊的處理該場景業(yè)務(wù)。
2、等待通知機(jī)制
如上的業(yè)務(wù)場景,如果線程A生成數(shù)據(jù)過程中,線程B一直在訪問數(shù)據(jù)容器,判斷該過程的數(shù)據(jù)是否已經(jīng)生成,則會造成資源浪費(fèi)。正常的流程應(yīng)該如圖,線程A和線程B同時啟動,線程A開始處理數(shù)據(jù)生成任務(wù),線程B嘗試獲取容器數(shù)據(jù),數(shù)據(jù)還沒過來,線程B則進(jìn)入等待狀態(tài),當(dāng)線程A的任務(wù)處理完成,則通知線程B去容器中獲取數(shù)據(jù),這樣基于線程等待和通知的機(jī)制來協(xié)作完成任務(wù)。
3、基礎(chǔ)方法
等待/通知機(jī)制的相關(guān)方法是Java中Object層級的基礎(chǔ)方法,任何對象都有該方法:
- notify:隨機(jī)通知一個在該對象上等待的線程,使其結(jié)束wait狀態(tài)返回;
- notifyAll:喚醒在該對象上所有等待的線程,進(jìn)入對象鎖爭搶隊列中;
- wait:線程進(jìn)入waiting等待狀態(tài),不會爭搶鎖對象,也可以設(shè)置等待時間;
線程的等待通知機(jī)制,就是基于這幾個基礎(chǔ)方法。
二、等待通知原理
1、基本原理
等待/通知機(jī)制,該模式下指線程A在不滿足任務(wù)執(zhí)行的情況下調(diào)用對象wait()方法進(jìn)入等待狀態(tài),線程B修改了線程A的執(zhí)行條件,并調(diào)用對象notify()或者notifyAll()方法,線程A收到通知后從wait狀態(tài)返回,進(jìn)而執(zhí)行后續(xù)操作。兩個線程通過基于對象提供的wait()/notify()/notifyAll()等方法完成等待和通知間交互,提高程序的可伸縮性。
2、實(shí)現(xiàn)案例
通過線程通信解決上述數(shù)據(jù)生成和存儲任務(wù)的解耦流程。
public class NotifyThread01 {static Object lock = new Object() ;static volatile List<String> dataList = new ArrayList<>();public static void main(String[] args) throws Exception {Thread saveThread = new Thread(new SaveData(),"SaveData");saveThread.start();TimeUnit.SECONDS.sleep(3);Thread dataThread = new Thread(new AnalyData(),"AnalyData");dataThread.start();}// 等待數(shù)據(jù)生成,保存static class SaveData implements Runnable {@Overridepublic void run() {synchronized (lock){while (dataList.size()==0){try {System.out.println(Thread.currentThread().getName()+"等待...");lock.wait();} catch (InterruptedException e) {e.printStackTrace();}}System.out.println("SaveData .."+ dataList.get(0)+dataList.get(1));}}}// 生成數(shù)據(jù),通知保存static class AnalyData implements Runnable {@Overridepublic void run() {synchronized (lock){dataList.add("hello,");dataList.add("java");lock.notify();System.out.println("AnalyData End...");}}} }注意:除了dataList滿足寫條件,還要在AnalyData線程執(zhí)行通知操作。
三、管道流通信
1、管道流簡介
基本概念
管道流主要用于在不同線程間直接傳送數(shù)據(jù),一個線程發(fā)送數(shù)據(jù)到輸出管道,另一個線程從輸入管道中讀取數(shù)據(jù),進(jìn)而實(shí)現(xiàn)不同線程間的通信。
實(shí)現(xiàn)分類
管道字節(jié)流:PipedInputStream和PipedOutputStream;
管道字符流:PipedWriter和PipedReader;
新IO管道流:Pipe.SinkChannel和Pipe.SourceChannel;
2、使用案例
public class NotifyThread02 {public static void main(String[] args) throws Exception {PipedInputStream pis = new PipedInputStream();PipedOutputStream pos = new PipedOutputStream();// 鏈接輸入流和輸出流pos.connect(pis);// 寫數(shù)據(jù)線程new Thread(new Runnable() {public void run() {BufferedReader br = new BufferedReader(new InputStreamReader(System.in));// 將從鍵盤讀取的數(shù)據(jù)寫入管道流PrintStream ps = new PrintStream(pos);while (true) {try {System.out.print(Thread.currentThread().getName());ps.println(br.readLine());Thread.sleep(1000);} catch (Exception e) {e.printStackTrace();}}}}, "輸入數(shù)據(jù)線程:").start();// 讀數(shù)據(jù)線程new Thread(new Runnable() {public void run() {BufferedReader br = new BufferedReader(new InputStreamReader(pis));while (true) {try {System.out.println(Thread.currentThread().getName() + br.readLine());} catch (IOException e) {e.printStackTrace();}}}}, "輸出數(shù)據(jù)線程:").start();} }寫線程向管道流寫入數(shù)據(jù),讀線程讀取數(shù)據(jù),完成基本通信流程。
四、生產(chǎn)消費(fèi)模式
1、業(yè)務(wù)場景
基于線程等待通知機(jī)制:實(shí)現(xiàn)工廠生產(chǎn)一件商品,通知商店賣出一件商品的業(yè)務(wù)流程。
2、代碼實(shí)現(xiàn)
public class NotifyThread03 {public static void main(String[] args) {Product product = new Product();ProductFactory productFactory = new ProductFactory(product);ProductShop productShop = new ProductShop(product);productFactory.start();productShop.start();} } // 產(chǎn)品 class Product {public String name ;public double price ;// 產(chǎn)品是否生產(chǎn)完畢,默認(rèn)沒有boolean flag ; } // 產(chǎn)品工廠:生產(chǎn) class ProductFactory extends Thread {Product product ;public ProductFactory (Product product){this.product = product;}@Overridepublic void run() {int i = 0 ;while (i < 20) {synchronized (product) {if (!product.flag){if (i%2 == 0){product.name = "鼠標(biāo)";product.price = 79.99;} else {product.name = "鍵盤";product.price = 89.99;}System.out.println("產(chǎn)品:"+product.name+"【價格:"+product.price+"】出廠...");product.flag = true ;i++;// 通知消費(fèi)者product.notifyAll();} else {try {// 進(jìn)入等待狀態(tài)product.wait();} catch (InterruptedException e) {e.printStackTrace();}}}}} } // 產(chǎn)品商店:銷售 class ProductShop extends Thread {Product product ;public ProductShop (Product product){this.product = product ;}@Overridepublic void run() {while (true) {synchronized (product) {if (product.flag == true ){System.out.println("產(chǎn)品:"+product.name+"【價格"+(product.price*2)+"】賣出...");product.flag = false ;product.notifyAll(); //喚醒生產(chǎn)者} else {try {product.wait();} catch (InterruptedException e) {e.printStackTrace();}}}}} }流程描述:ProductFactory生成一件商品,通知商店售賣,通過flag標(biāo)識判斷控制是否進(jìn)入等待狀態(tài),商店賣出商品后,再次通知工廠生產(chǎn)商品。
五、源代碼地址
GitHub·地址 https://github.com/cicadasmile/java-base-parent GitEE·地址 https://gitee.com/cicadasmile/java-base-parent| 01 | Java并發(fā):線程的創(chuàng)建方式,狀態(tài)周期管理 |
| 02 | Java并發(fā):線程核心機(jī)制,基礎(chǔ)概念擴(kuò)展 |
| 03 | Java并發(fā):多線程并發(fā)訪問,同步控制 |
總結(jié)
以上是生活随笔為你收集整理的Java并发编程(04):线程间通信,等待/通知机制的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 快乐的冬至
- 下一篇: ABAP中的F4帮助怎么用