Java多線程編程概述
Java多線程的安全問題
Java多線程同步
- Java多線程同步機(jī)制簡介
- Java多線程鎖
- Java多線程同步代碼塊Synchronized
- Java多線程同步方法
- Java多線程臟讀
- Java多線程出現(xiàn)異常會自動釋放鎖
- Java多線程死鎖
- Java volatile關(guān)鍵字的作用
- Java多線程之volatile非原子性
- Java原子類自增自減操作
- Java CAS多線程
- Java原子變量
Java多線程間的通信
- Java線程間的通信方式
- Java多線程的等待通知機(jī)制
- Java生產(chǎn)者消費者模式
- Java中利用管道實現(xiàn)線程間的通訊
- Java線程中join方法
- Java ThreadLocal使用
Java線程Lock
- Java中鎖的可重入性
- Java ReentrantLock使用
- ReentrantLock可重入性
- Java多線程：lockInterruptibly()方法
- Java多線程：tryLock()方法
- Java多線程：newCondition()方法
- Java公平鎖與非公平鎖
- Lock鎖的常用方法
- Java多線程讀寫鎖
Java多線程管理
- Java線程組
- Java多線程捕獲異常處理
- Hook鉤子線程注入
- Java線程池是什么
- 多線程JDK線程池
- Java線程池的底層實現(xiàn)
- Java線程池的拒絕策略
- Java線程池ThreadFactory
- Java監(jiān)控線程池
- Java線程池擴(kuò)展
- Java線程池的大小與線程池死鎖
- Java線程池異常處理
- Java ForkJoinPool線程池
保障線程安全的設(shè)計技術(shù)
Java鎖的優(yōu)化及注意事項
Java多線程集合
【Java多線程】單例模式與多線程
Java多線程的等待通知機(jī)制
什么是等待通知機(jī)制
在單線程編程中,要執(zhí)行的操作需要滿足一定的條件才能執(zhí)行,可以把這個操作放在if語句塊中。
在多線程編程中,可能A線程的條件沒有滿足只是暫時的, 稍后其他的線程B可能會更新條件使得A線程的條件得到滿足. 可以將A線程暫停,直到它的條件得到滿足后再將A線程喚醒.它的偽代碼：
atomics{ //原子操作
while( 條件不成立 ){
等待
}
當(dāng)前線程被喚醒條件滿足后,繼續(xù)執(zhí)行下面的操作
}
等待/通知機(jī)制的實現(xiàn)
Object類中的wait()方法可以使執(zhí)行當(dāng)前代碼的線程等待,暫停執(zhí)行,直到接到通知或被中斷為止。
注意:
● wait()方法只能 在同步代碼塊中由鎖對象調(diào)用。
● 調(diào)用wait()方法,當(dāng)前線程會釋放鎖。
其偽代碼如下:
//在調(diào)用wait()方法前獲得對象的內(nèi)部鎖
synchronized( 鎖對象 ){
while( 條件不成立 ){
//通過鎖對象調(diào)用 wait()方法暫停線程,會釋放鎖對象
鎖對象.wait();
}
//線程的條件滿足了繼續(xù)向下執(zhí)行
}
Object類的notify()可以喚醒線程,該方法也必須在同步代碼塊中由鎖對象調(diào)用. 沒有使用鎖對象調(diào)用 wait()/notify()會拋出IlegalMonitorStateExeption異常. 如果有多個等待的線程,notify()方法只能喚醒其中的一個. 在同步代碼塊中調(diào)用notify()方法后,并不會立即釋放鎖對象,需要等當(dāng)前同步代碼塊執(zhí)行完后才會釋放鎖對象,一般將notify()方法放在同步代碼塊的最后. 它的偽代碼如下:
synchronized( 鎖對象 ){
//執(zhí)行修改保護(hù)條件 的代碼
//喚醒其他線程
鎖對象.notify();
}
package com.wkcto.wait;
/**
* 需要通過notify()喚醒等待的線程
* 北京動力節(jié)點老崔
*/
public class Test03 {
public static void main(String[] args) throws InterruptedException {
String lock = "wkcto"; //定義一個字符串作為鎖對象
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
synchronized (lock) {
System.out.println("線程1開始等待: " + System.currentTimeMillis());
try {
lock.wait(); //線程等待,會釋放鎖對象,當(dāng)前線程轉(zhuǎn)入blocked阻塞狀態(tài)
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("線程1結(jié)束等待:" + System.currentTimeMillis());
}
}
});
//定義第二個線程,在第二個線程中喚醒第一個線程
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
//notify()方法也需要在同步代碼塊中,由鎖對象調(diào)用
synchronized (lock){
System.out.println("線程2開始喚醒 : " + System.currentTimeMillis());
lock.notify(); //喚醒在lock鎖對象上等待的某一個線程
System.out.println("線程2結(jié)束喚醒 : " + System.currentTimeMillis());
}
}
});
t1.start(); //開啟t1線程,t1線程等待
Thread.sleep(3000); //main線程睡眠3秒,確保t1入睡
t2.start(); //t1線程開啟3秒后,再開啟t2線程喚醒t1線程
}
}
notify()方法后不會立即釋放鎖對象
package com.wkcto.wait;
import java.util.ArrayList;
import java.util.List;
/**
* notify()不會立即釋放鎖對象
* 北京動力節(jié)點老崔
*/
public class Test04 {
public static void main(String[] args) throws InterruptedException {
//定義一個List集合存儲String數(shù)據(jù)
List<String> list = new ArrayList<>();
//定義第一個線程,當(dāng)list集合中元素的數(shù)量不等于5時線程等待
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
synchronized (list){
if ( list.size() != 5 ){
System.out.println("線程1開始等待: " + System.currentTimeMillis());
try {
list.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("線程1被喚醒:" + System.currentTimeMillis());
}
}
}
});
//定義第二個線程,向list集合中添加元素
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
synchronized (list){
for (int i = 0; i < 10; i++) {
list.add("data--" + i);
System.out.println("線程2添加了第" + (i+1) + "個數(shù)據(jù)");
//判斷元素的數(shù)量是否滿足喚醒線程1
if (list.size() == 5 ){
list.notify(); //喚醒 線程, 不會立即釋放鎖對象,需要等到當(dāng)前同步代碼塊都執(zhí)行完后才能釋放鎖對象
System.out.println("線程2已經(jīng)發(fā)現(xiàn)喚醒通知");
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
});
t1.start();
//為了確保t2在t1之后開啟,即讓t1線程先睡眠
Thread.sleep(500);
t2.start();
}
}
interrupt()方法會中斷wait()
當(dāng)線程處于wait()等待狀態(tài)時, 調(diào)用線程對象的interrupt()方法會中斷線程的等待狀態(tài), 會產(chǎn)生InterruptedException異常。
package com.wkcto.wait;
/**
* Interrupt()會中斷線程的wait()等待
* 北京動力節(jié)點老崔
*/
public class Test05 {
public static void main(String[] args) throws InterruptedException {
SubThread t = new SubThread();
t.start();
Thread.sleep(2000); //主線程睡眠2秒, 確保子線程處于Wait等待狀態(tài)
t.interrupt();
}
private static final Object LOCK = new Object(); //定義常量作為鎖對象
static class SubThread extends Thread{
@Override
public void run() {
synchronized (LOCK){
try {
System.out.println("begin wait...");
LOCK.wait();
System.out.println("end wait..");
} catch (InterruptedException e) {
System.out.println("wait等待被中斷了****");
}
}
}
}
}
notify()與notifyAll()
notify()一次只能喚醒一個線程,如果有多個等待的線程,只能隨機(jī)喚醒其中的某一個; 想要喚醒所有等待線程,需要調(diào)用notifyAll()。
package com.wkcto.wait;
/**
* notify()與notifyAll()
* 北京動力節(jié)點老崔
*/
public class Test06 {
public static void main(String[] args) throws InterruptedException {
Object lock = new Object(); //定義一個對象作為子線程的鎖對象
SubThread t1 = new SubThread(lock);
SubThread t2 = new SubThread(lock);
SubThread t3 = new SubThread(lock);
t1.setName("t1");
t2.setName("t2");
t3.setName("t3");
t1.start();
t2.start();
t3.start();
Thread.sleep(2000);
//調(diào)用notify()喚醒 子線程
synchronized (lock){
// lock.notify(); //調(diào)用一次notify()只能喚醒其中的一個線程,其他等待的線程依然處于等待狀態(tài),對于處于等待狀態(tài)的線程來說,錯過了通知信號,這種現(xiàn)象也稱為信號丟失
lock.notifyAll(); //喚醒所有的線程
}
}
static class SubThread extends Thread{
private Object lock; //定義實例變量作為鎖對象
public SubThread(Object lock) {
this.lock = lock;
}
@Override
public void run() {
synchronized (lock){
try {
System.out.println(Thread.currentThread().getName() + " -- begin wait...");
lock.wait();
System.out.println( Thread.currentThread().getName() + " -- end wait...");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
wait(long)的使用
wait(long)帶有l(wèi)ong類型參數(shù)的wait()等待,如果在參數(shù)指定的時間內(nèi)沒有被喚醒,超時后會自動喚醒。
package com.wkcto.wait;
/**
* wait(long)
* 北京動力節(jié)點老崔
*/
public class Test07 {
public static void main(String[] args) {
final Object obj = new Object();
Thread t = new Thread(new Runnable() {
@Override
public void run() {
synchronized ( obj ){
try {
System.out.println("thread begin wait");
obj.wait(5000); //如果5000毫秒內(nèi)沒有被喚醒 ,會自動喚醒
System.out.println("end wait....");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
t.start();
}
}
通知過早
線程wait()等待后,可以調(diào)用notify()喚醒線程, 如果notify()喚醒的過早,在等待之前就調(diào)用了notify()可能會打亂程序正常的運(yùn)行邏輯。
package com.wkcto.wait;
/**
* notify()通知過早
* 北京動力節(jié)點老崔
*/
public class Test08 {
public static void main(String[] args) {
final Object Lock = new Object(); //定義對象作為鎖對象
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
synchronized (Lock){
try {
System.out.println("begin wait");
Lock.wait();
System.out.println("wait end...");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
synchronized (Lock){
System.out.println("begin notify");
Lock.notify();;
System.out.println("end nofity");
}
}
});
//如果先開啟t1,再開啟t2線程,大多數(shù)情況下, t1先等待,t1再把t1喚醒
// t1.start();
// t2.start();
//如果先開啟t2通知線程,再開啟t1等待線程,可能會出現(xiàn)t1線程等待沒有收到通知的情況,
t2.start();
t1.start();
}
}
package com.wkcto.wait;
/**
* notify()通知過早, 就不讓線程等待了
* 北京動力節(jié)點老崔
*/
public class Test09 {
static boolean isFirst = true; //定義靜態(tài)變量作為是否第一個運(yùn)行的線程標(biāo)志
public static void main(String[] args) {
final Object Lock = new Object(); //定義對象作為鎖對象
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
synchronized (Lock){
while ( isFirst ) { //當(dāng)線程是第一個開啟的線程就等待
try {
System.out.println("begin wait");
Lock.wait();
System.out.println("wait end...");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
});
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
synchronized (Lock){
System.out.println("begin notify");
Lock.notify();;
System.out.println("end nofity");
isFirst = false; //通知后,就把第一個線程標(biāo)志修改為false
}
}
});
//如果先開啟t1,再開啟t2線程,大多數(shù)情況下, t1先等待,t1再把t1喚醒
// t1.start();
// t2.start();
//如果先開啟t2通知線程,再開啟t1等待線程,可能會出現(xiàn)t1線程等待沒有收到通知的情況,
t2.start();
t1.start();
//實際上,調(diào)用start()就是告訴線程調(diào)度器,當(dāng)前線程準(zhǔn)備就緒,線程調(diào)度器在什么時候開啟這個線程不確定,即調(diào)用start()方法的順序,并不一定就是線程實際開啟的順序.
//在當(dāng)前示例中,t1等待后讓t2線程喚醒 , 如果t2線程先喚醒了,就不讓t1線程等待了
}
}
wait等待條件發(fā)生了變化
在使用wait/nofity模式時,注意wait條件發(fā)生了變化,也可能會造成邏輯的混亂。
package com.wkcto.wait;
import java.util.ArrayList;
import java.util.List;
/**
* wait條件發(fā)生變化
* 定義一個集合
* 定義一個線程向集合中添加數(shù)據(jù),添加完數(shù)據(jù)后通知另外的線程從集合中取數(shù)據(jù)
* 定義一個線程從集合中取數(shù)據(jù),如果集合中沒有數(shù)據(jù)就等待
* 北京動力節(jié)點老崔
*/
public class Test10 {
public static void main(String[] args) {
//定義添加數(shù)據(jù)的線程對象
ThreadAdd threadAdd = new ThreadAdd();
//定義取數(shù)據(jù)的線程對象
ThreadSubtract threadSubtract = new ThreadSubtract();
threadSubtract.setName("subtract 1 ");
//測試一: 先開啟添加數(shù)據(jù)的線程,再開啟一個取數(shù)據(jù)的線程,大多數(shù)情況下會正常取數(shù)據(jù)
// threadAdd.start();
// threadSubtract.start();
//測試二: 先開啟取數(shù)據(jù)的線程,再開啟添加數(shù)據(jù)的線程, 取數(shù)據(jù)的線程會先等待,等到添加數(shù)據(jù)之后 ,再取數(shù)據(jù)
// threadSubtract.start();
// threadAdd.start();
//測試三: 開啟兩個 取數(shù)據(jù)的線程,再開啟添加數(shù)據(jù)的線程
ThreadSubtract threadSubtract2 = new ThreadSubtract();
threadSubtract2.setName("subtract 2 ");
threadSubtract.start();
threadSubtract2.start();
threadAdd.start();
/*
某一次執(zhí)行結(jié)果如下:
subtract 1 begin wait....
subtract 2 從集合中取了data后,集合中數(shù)據(jù)的數(shù)量:0
subtract 1 end wait..
Exception in thread "subtract 1 " java.lang.IndexOutOfBoundsException:
分析可能的執(zhí)行順序:
threadSubtract線程先啟動, 取數(shù)據(jù)時,集合中沒有數(shù)據(jù),wait()等待
threadAdd線程獲得CPU執(zhí)行權(quán), 添加數(shù)據(jù) , 把threadSubtract線程喚醒,
threadSubtract2線程開啟后獲得CPU執(zhí)行權(quán), 正常取數(shù)據(jù)
threadSubtract線程獲得CPU執(zhí)行權(quán), 打印 end wait..., 然后再執(zhí)行l(wèi)ist.remove(0)取數(shù)據(jù)時,現(xiàn)在list集合中已經(jīng)沒有數(shù)據(jù)了, 這時會產(chǎn)生java.lang.IndexOutOfBoundsException異常
出現(xiàn)異常的原因是: 向list集合中添加了一個數(shù)據(jù),remove()了兩次
如何解決?
當(dāng)?shù)却木€程被喚醒后, 再判斷一次集合中是否有數(shù)據(jù)可取. 即需要把sutract()方法中的if判斷改為while
*/
}
//1)定義List集合
static List list = new ArrayList<>();
//2)定義方法從集合中取數(shù)據(jù)
public static void subtract(){
synchronized (list) {
// if (list.size() == 0) {
while (list.size() == 0) {
try {
System.out.println(Thread.currentThread().getName() + " begin wait....");
list.wait(); //等待
System.out.println(Thread.currentThread().getName() + " end wait..");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
Object data = list.remove(0); //從集合中取出一個數(shù)據(jù)
System.out.println( Thread.currentThread().getName() + "從集合中取了" + data + "后,集合中數(shù)據(jù)的數(shù)量:" + list.size());
}
}
//3)定義方法向集合中添加數(shù)據(jù)后,通知等待的線程取數(shù)據(jù)
public static void add(){
synchronized (list){
list.add("data");
System.out.println( Thread.currentThread().getName() + "存儲了一個數(shù)據(jù)");
list.notifyAll();
}
}
//4)定義線程類調(diào)用add()取數(shù)據(jù)的方法
static class ThreadAdd extends Thread{
@Override
public void run() {
add();
}
}
//定義線程類調(diào)用subtract()方法
static class ThreadSubtract extends Thread{
@Override
public void run() {
subtract();
}
}
}
