一、使用synchronize以及wait()汞幢、notify() /notifyAll()
- 1.wait()的作用是讓當(dāng)前線程進(jìn)入等待狀態(tài)曹抬,同時(shí),wait()也會(huì)讓當(dāng)前線程釋放它所持有的鎖急鳄“瘢“直到其他線程調(diào)用此對(duì)象的 notify() 方法或 notifyAll() 方法”,當(dāng)前線程被喚醒(進(jìn)入“就緒狀態(tài)”)
- 2.notify()和notifyAll()的作用疾宏,則是喚醒當(dāng)前對(duì)象上的等待線程张足;notify()是喚醒單個(gè)線程,而notifyAll()是喚醒所有的線程坎藐。
- 3.wait(long timeout)讓當(dāng)前線程處于“等待(阻塞)狀態(tài)”为牍,“直到其他線程調(diào)用此對(duì)象的notify()方法或 notifyAll() 方法,或者超過指定的時(shí)間量”岩馍,當(dāng)前線程被喚醒(進(jìn)入“就緒狀態(tài)”)碉咆。
/**
* 使用synchronize以及wait()、notify() /notifyAll()
*/
public class ShareDataV1 {
/**
* 原子計(jì)數(shù)
*/
public static AtomicInteger atomicInteger = new AtomicInteger();
/**
* 標(biāo)示
*/
public volatile boolean flag = true;
/**
* 生產(chǎn)隊(duì)列最大容量
*/
public static final int MAX_COUNT = 10;
/**
* 生產(chǎn)隊(duì)列容器
*/
public static final List<Integer> pool = new ArrayList<>();
/**
* 生產(chǎn)
*/
public void produce() {
// 判斷蛀恩,干活疫铜,通知
while (flag) {
try {
// 每隔 1000 毫秒生產(chǎn)一個(gè)商品
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 這里有一個(gè)鎖
synchronized (pool) {
//池子滿了,生產(chǎn)者停止生產(chǎn)
//埋個(gè)坑双谆,這里用的if
//TODO 判斷
while (pool.size() == MAX_COUNT) {
try {
System.out.println("pool is full, wating...");
// TODO 線程阻塞
pool.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//干活
pool.add(atomicInteger.incrementAndGet());
System.out.println("produce number:" + atomicInteger.get() + "\t" + "current size:" + pool.size());
//通知
// notify() 方法隨機(jī)喚醒對(duì)象的等待池中的一個(gè)線程壳咕,進(jìn)入鎖池席揽;
// notifyAll() 喚醒對(duì)象的等待池中的所有線程,進(jìn)入鎖池谓厘。
pool.notifyAll();
}
}
}
/**
* 消費(fèi)
*/
public void consumue() {
// 判斷幌羞,干活,通知
while (flag) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 這里有一個(gè)鎖
synchronized (pool) {
//池子滿了竟稳,生產(chǎn)者停止生產(chǎn)
//埋個(gè)坑属桦,這里用的if
//TODO 判斷
while (pool.size() == 0) {
try {
System.out.println("pool is empty, wating...");
pool.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//干活
int temp = pool.get(0);
pool.remove(0);
System.out.println("cousume number:" + temp + "\t" + "current size:" + pool.size());
//通知
pool.notifyAll();
}
}
}
/**
* 停止
*/
public void stop() {
flag = false;
}
public static void main(String[] args) {
ShareDataV1 shareDataV1 = new ShareDataV1();
// 開啟生產(chǎn)線程
new Thread(() -> {
shareDataV1.produce();
}, "AAA").start();
// 開啟消費(fèi)線程
new Thread(() -> {
shareDataV1.consumue();
}, "BBB").start();
// 開啟生產(chǎn)線程
new Thread(() -> {
shareDataV1.produce();
}, "CCC").start();
// 開啟消費(fèi)線程
new Thread(() -> {
shareDataV1.consumue();
}, "DDD").start();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
shareDataV1.stop();
}
}
二、使用Lock他爸,Condition的await和signal方法
- JUC包下的鎖Lock替代synchronize關(guān)鍵字地啰。await方法代替wait,signal代替notifyall讲逛。
下面這個(gè)demo實(shí)現(xiàn)了pool的大小為1的生產(chǎn)者消費(fèi)者模型亏吝。
class ShareData {
private int number = 0;
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
public void increment() throws Exception {
lock.lock();
try {
while (number != 0) {
//等待,不能生產(chǎn)
condition.await();
}
number++;
System.out.println(Thread.currentThread().getName() + "\t" + number);
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void decrement() throws Exception {
lock.lock();
try {
while (number == 0) {
//等待盏混,不能消費(fèi)
condition.await();
}
number--;
System.out.println(Thread.currentThread().getName() + "\t" + number);
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
public class ProducerConsumer_V2{
public static void main(String[] args) {
ShareData shareData = new ShareData();
new Thread(()->{
for (int i = 0; i < 5; i++) {
try {
shareData.increment();
} catch (Exception e) {
e.printStackTrace();
}
}
}, "AA").start();
new Thread(()->{
for (int i = 0; i < 5; i++) {
try {
shareData.decrement();
} catch (Exception e) {
e.printStackTrace();
}
}
}, "BB").start();
new Thread(()->{
for (int i = 0; i < 5; i++) {
try {
shareData.increment();
} catch (Exception e) {
e.printStackTrace();
}
}
}, "CC").start();
new Thread(()->{
for (int i = 0; i < 5; i++) {
try {
shareData.decrement();
} catch (Exception e) {
e.printStackTrace();
}
}
}, "DD").start();
}
}
三蔚鸥、下面是使用阻塞隊(duì)列實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模式:
- 當(dāng)阻塞隊(duì)列為空時(shí),從阻塞隊(duì)列中取數(shù)據(jù)的操作會(huì)被阻塞许赃。
- 當(dāng)阻塞隊(duì)列為滿時(shí)止喷,往阻塞隊(duì)列中添加數(shù)據(jù)的操作會(huì)被阻塞。
LinkedBlockingQueue
- 基于單向鏈表的阻塞隊(duì)列實(shí)現(xiàn)混聊,在初始化LinkedBlockingQueue的時(shí)候可以指定對(duì)立的大小弹谁,也可以不指定,默認(rèn)類似一個(gè)無限大小的容量(Integer.MAX_VALUE)句喜,不指隊(duì)列容量大小也是會(huì)有風(fēng)險(xiǎn)的预愤,一旦數(shù)據(jù)生產(chǎn)速度大于消費(fèi)速度,系統(tǒng)內(nèi)存將有可能被消耗殆盡咳胃,因此要謹(jǐn)慎操作植康。另外LinkedBlockingQueue中用于阻塞生產(chǎn)者、消費(fèi)者的鎖是兩個(gè)(鎖分離)展懈,因此生產(chǎn)與消費(fèi)是可以同時(shí)進(jìn)行的销睁。
/**
* 下面是使用阻塞隊(duì)列實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模式:
*/
public class ShareDataV3 {
/**
* 阻塞隊(duì)列容量
*/
private static final int MAX_CAPACITY = 10;
/**
* 阻塞隊(duì)列
*/
private static BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>(MAX_CAPACITY);
/**
* 標(biāo)示
*/
private volatile boolean FLAG = true;
/**
* 原子計(jì)數(shù)
*/
private AtomicInteger atomicInteger = new AtomicInteger();
/**
* 生產(chǎn)
*
* @throws InterruptedException
*/
public void produce() throws InterruptedException {
// 開啟可控的死循環(huán) 不停的生產(chǎn)
while (FLAG) {
// 向隊(duì)列中添加元素
boolean retvalue = blockingQueue.offer(atomicInteger.incrementAndGet(), 2, TimeUnit.SECONDS);
// 插入元素是否成功
if (retvalue == true) {
System.out.println(Thread.currentThread().getName() + "\t 插入隊(duì)列" + atomicInteger.get() + "成功" + "資源隊(duì)列大小= " + blockingQueue.size());
} else {
System.out.println(Thread.currentThread().getName() + "\t 插入隊(duì)列" + atomicInteger.get() + "失敗" + "資源隊(duì)列大小= " + blockingQueue.size());
}
// 線程睡眠
TimeUnit.SECONDS.sleep(1);
}
System.out.println(Thread.currentThread().getName() + "FLAG變?yōu)閒lase,生產(chǎn)停止");
}
/**
* 消費(fèi)
*
* @throws InterruptedException
*/
public void consume() throws InterruptedException {
Integer result = null;
while (true) {
// 從隊(duì)列中獲取元素
result = blockingQueue.poll(2, TimeUnit.SECONDS);
// 獲取結(jié)果判斷
if (null == result) {
System.out.println("超過兩秒沒有取道數(shù)據(jù)存崖,消費(fèi)者即將退出");
return;
}
// 打印獲取結(jié)果
System.out.println(Thread.currentThread().getName() + "\t 消費(fèi)" + result + "成功" + "\t\t" + "資源隊(duì)列大小= " + blockingQueue.size());
Thread.sleep(1500);
}
}
/**
* 停止
*/
public void stop() {
this.FLAG = false;
}
public static void main(String[] args) {
ShareDataV3 v3 = new ShareDataV3();
// 開啟一個(gè)線程 執(zhí)行生產(chǎn)
new Thread(() -> {
try {
v3.produce();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "AAA").start();
// 開啟一個(gè)線程 執(zhí)行消費(fèi)
new Thread(() -> {
try {
v3.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "BBB").start();
// 開啟一個(gè)線程 執(zhí)行消費(fèi)
new Thread(() -> {
try {
v3.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "CCC").start();
try {
// 測試 主線程睡眠5秒
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
v3.stop();
}
}