一花墩、使用synchronize以及wait()雳刺、notify() /notifyAll()
package com.zhb.juc;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 使用synchronize wait notify/notifyall實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模式
*/
class ShareDataV1 {
public static AtomicInteger atomicInteger = new AtomicInteger();
public volatile boolean flag = true;
public static final int MAX_COUNT = 10;
public static final List<Integer> pool = new ArrayList<>();
public void produce() {
// 判斷盔性，干活仲锄，通知
while (flag) {
// 每隔 1000 毫秒生產(chǎn)一個(gè)商品
try {
Thread.sleep(100);
} catch (InterruptedException e) {
}
synchronized (pool) {
//池子滿了劲妙，生產(chǎn)者停止生產(chǎn)
//埋個(gè)坑，這里用的if
//TODO 判斷
if (pool.size() == MAX_COUNT) {
try {
System.out.println("pool is full, wating...");
pool.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//干活
pool.add(atomicInteger.incrementAndGet());
System.out.println("produce number:" + atomicInteger.get() + "\t" + "current size:" + pool.size());
//通知
pool.notifyAll();
}
}
}
public void consumue() {
// 判斷儒喊，干活镣奋，通知
while (flag) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
synchronized (pool) {
//池子滿了，生產(chǎn)者停止生產(chǎn)
//埋個(gè)坑怀愧，這里用的if
//TODO 判斷
if (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 class ProducerConsumer_V1 {
public static void main(String[] args) {
ShareDataV1 shareDataV1 = new ShareDataV1();
new Thread(() -> {
shareDataV1.produce();
}, "AAA").start();
new Thread(() -> {
shareDataV1.consumue();
}, "BBB").start();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
shareDataV1.stop();
}
}
上面的程序在只有兩個(gè)線程時(shí)（一個(gè)生產(chǎn)者侨颈，一個(gè)消費(fèi)者）可以正常工作。打印的log如下：
produce number:1 current size:1
produce number:2 current size:2
produce number:3 current size:3
produce number:4 current size:4
produce number:5 current size:5
produce number:6 current size:6
produce number:7 current size:7
produce number:8 current size:8
produce number:9 current size:9
cousume number:1 current size:8
produce number:10 current size:9
produce number:11 current size:10
pool is full, wating...
cousume number:2 current size:9
produce number:12 current size:10
pool is full, wating...
cousume number:3 current size:9
produce number:13 current size:10
pool is full, wating...
cousume number:4 current size:9
produce number:14 current size:10
pool is full, wating...
cousume number:5 current size:9
produce number:15 current size:10
Process finished with exit code 0
但是我們把生產(chǎn)者和消費(fèi)者線程擴(kuò)展至多個(gè)芯义。就出錯(cuò)了哈垢。例如再增加CCC和DDD線程分別生產(chǎn)和消費(fèi)。只改動(dòng)了main方法：
public class ProducerConsumer_V1 {
public static void main(String[] args) {
ShareDataV1 shareDataV1 = new ShareDataV1();
new Thread(() -> {
shareDataV1.produce();
}, "AAA").start();
new Thread(() -> {
shareDataV1.consumue();
}, "BBB").start();
new Thread(() -> {
shareDataV1.produce();
}, "CCC").start();
new Thread(() -> {
shareDataV1.consumue();
}, "DDD").start();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
shareDataV1.stop();
}
}
輸出的log如下：
produce number:1 current size:1
produce number:2 current size:2
produce number:3 current size:3
produce number:4 current size:4
produce number:5 current size:5
produce number:6 current size:6
produce number:7 current size:7
produce number:8 current size:8
produce number:9 current size:9
produce number:10 current size:10
pool is full, wating...
pool is full, wating...
cousume number:1 current size:9
cousume number:2 current size:8
produce number:11 current size:9
produce number:12 current size:10
pool is full, wating...
pool is full, wating...
cousume number:3 current size:9
produce number:13 current size:10
produce number:14 current size:11
cousume number:4 current size:10
pool is full, wating...
pool is full, wating...
cousume number:5 current size:9
produce number:15 current size:10
produce number:16 current size:11
cousume number:6 current size:10
pool is full, wating...
pool is full, wating...
cousume number:7 current size:9
produce number:17 current size:10
produce number:18 current size:11
cousume number:8 current size:10
pool is full, wating...
pool is full, wating...
cousume number:9 current size:9
produce number:19 current size:10
produce number:20 current size:11
cousume number:10 current size:10
pool is full, wating...
pool is full, wating...
cousume number:11 current size:9
produce number:21 current size:10
produce number:22 current size:11
Process finished with exit code 0
我們看到current size 能到11了扛拨。這肯定出錯(cuò)了耘分。因?yàn)槲覀円髉ool的最大容量為10。出現(xiàn)這個(gè)情況的原因是在多線程的環(huán)境下绑警，要防止虛假喚醒求泰。即判斷條件不能用if，而是用while计盒。接下來(lái)我們修改上面//TODO部分的代碼拜秧，把if改成while再來(lái)測(cè)試。最終版正確的代碼如下：
package com.zhb.juc;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 使用synchronize wait notify/notifyall實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模式
*/
class ShareDataV1 {
public static AtomicInteger atomicInteger = new AtomicInteger();
public volatile boolean flag = true;
public static final int MAX_COUNT = 10;
public static final List<Integer> pool = new ArrayList<>();
public void produce() {
// 判斷章郁，干活枉氮，通知
while (flag) {
// 每隔 1000 毫秒生產(chǎn)一個(gè)商品
try {
Thread.sleep(100);
} catch (InterruptedException e) {
}
synchronized (pool) {
//池子滿了，生產(chǎn)者停止生產(chǎn)
//埋個(gè)坑暖庄，這里用的if
//TODO 判斷
while (pool.size() == MAX_COUNT) {
try {
System.out.println("pool is full, wating...");
pool.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//干活
pool.add(atomicInteger.incrementAndGet());
System.out.println("produce number:" + atomicInteger.get() + "\t" + "current size:" + pool.size());
//通知
pool.notifyAll();
}
}
}
public void consumue() {
// 判斷聊替，干活，通知
while (flag) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
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 class ProducerConsumer_V1 {
public static void main(String[] args) {
ShareDataV1 shareDataV1 = new ShareDataV1();
new Thread(() -> {
shareDataV1.produce();
}, "AAA").start();
new Thread(() -> {
shareDataV1.consumue();
}, "BBB").start();
new Thread(() -> {
shareDataV1.produce();
}, "CCC").start();
new Thread(() -> {
shareDataV1.consumue();
}, "DDD").start();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
shareDataV1.stop();
}
}
輸出結(jié)果：
produce number:1 current size:1
produce number:2 current size:2
produce number:3 current size:3
produce number:4 current size:4
produce number:5 current size:5
produce number:6 current size:6
produce number:7 current size:7
produce number:8 current size:8
produce number:9 current size:9
produce number:10 current size:10
pool is full, wating...
pool is full, wating...
cousume number:1 current size:9
produce number:11 current size:10
pool is full, wating...
cousume number:2 current size:9
produce number:12 current size:10
pool is full, wating...
pool is full, wating...
cousume number:3 current size:9
produce number:13 current size:10
pool is full, wating...
cousume number:4 current size:9
produce number:14 current size:10
pool is full, wating...
pool is full, wating...
cousume number:5 current size:9
produce number:15 current size:10
pool is full, wating...
cousume number:6 current size:9
produce number:16 current size:10
pool is full, wating...
pool is full, wating...
cousume number:7 current size:9
produce number:17 current size:10
pool is full, wating...
cousume number:8 current size:9
produce number:18 current size:10
pool is full, wating...
pool is full, wating...
cousume number:9 current size:9
produce number:19 current size:10
pool is full, wating...
cousume number:10 current size:9
produce number:20 current size:10
cousume number:11 current size:9
Process finished with exit code 0
二、使用Lock肩钠，Condition的await和signal方法
JUC包下的鎖Lock替代synchronize關(guān)鍵字泣港。await方法代替wait暂殖，signal代替notifyall。
下面這個(gè)demo實(shí)現(xiàn)了pool的大小為1的生產(chǎn)者消費(fèi)者模型当纱。
package com.zhb.juc;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
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ì)列
首先談?wù)勛枞?duì)列：
當(dāng)阻塞隊(duì)列為空時(shí)晨横，從阻塞隊(duì)列中取數(shù)據(jù)的操作會(huì)被阻塞。
當(dāng)阻塞隊(duì)列為滿時(shí)箫柳，往阻塞隊(duì)列中添加數(shù)據(jù)的操作會(huì)被阻塞手形。
JDK中的七大阻塞隊(duì)列
| 阻塞隊(duì)列名稱 | 說(shuō)明 |
|---|---|
| ArrayBlockingQueue | 一個(gè)由數(shù)組結(jié)構(gòu)組成的有界阻塞隊(duì)列。 |
| LinkedBlockingQueue | 一個(gè)由鏈表結(jié)構(gòu)組成的有界阻塞隊(duì)列悯恍。 |
| PriorityBlockingQueue | 一個(gè)支持優(yōu)先級(jí)排序的無(wú)界阻塞隊(duì)列库糠。 |
| DelayQueue | 一個(gè)使用優(yōu)先級(jí)隊(duì)列實(shí)現(xiàn)的無(wú)界阻塞隊(duì)列。 |
| SynchronousQueue | 一個(gè)不存儲(chǔ)元素的阻塞隊(duì)列涮毫。 |
| LinkedTransferQueue | 一個(gè)由鏈表結(jié)構(gòu)組成的無(wú)界阻塞隊(duì)列曼玩。 |
| LinkedBlockingDeque | 一個(gè)由鏈表結(jié)構(gòu)組成的雙向阻塞隊(duì)列。 |
ArrayBlockingQueue:
基于數(shù)組的阻塞隊(duì)列實(shí)現(xiàn)窒百，其內(nèi)部維護(hù)一個(gè)定長(zhǎng)的數(shù)組黍判，用于存儲(chǔ)隊(duì)列元素。線程阻塞的實(shí)現(xiàn)是通過(guò)ReentrantLock來(lái)完成的篙梢，數(shù)據(jù)的插入與取出共用同一個(gè)鎖顷帖，因此ArrayBlockingQueue并不能實(shí)現(xiàn)生產(chǎn)、消費(fèi)同時(shí)進(jìn)行渤滞。而且在創(chuàng)建ArrayBlockingQueue時(shí)贬墩，我們還可以控制對(duì)象的內(nèi)部鎖是否采用公平鎖，默認(rèn)采用非公平鎖妄呕。
LinkedBlockingQueue:
基于單向鏈表的阻塞隊(duì)列實(shí)現(xiàn)陶舞，在初始化LinkedBlockingQueue的時(shí)候可以指定對(duì)立的大小，也可以不指定绪励，默認(rèn)類似一個(gè)無(wú)限大小的容量（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)行的。
參考：http://www.reibang.com/p/f4eefb069e27
下面是使用阻塞隊(duì)列實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模式：
package com.zhb.juc;
/*
使用阻塞隊(duì)列實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模型
*/
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
//資源類
class ShareDataV3{
private static final int MAX_CAPACITY = 10; //阻塞隊(duì)列容量
private static BlockingQueue<Integer> blockingQueue= new ArrayBlockingQueue<>(MAX_CAPACITY); //阻塞隊(duì)列
private volatile boolean FLAG = true;
private AtomicInteger atomicInteger = new AtomicInteger();
public void produce() throws InterruptedException {
while (FLAG){
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)停止");
}
public void consume() throws InterruptedException {
Integer result = null;
while (true){
result = blockingQueue.poll(2, TimeUnit.SECONDS);
if (null==result){
System.out.println("超過(guò)兩秒沒(méi)有取道數(shù)據(jù)烙丛，消費(fèi)者即將退出");
return;
}
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 class ProducerConsumer_V3 {
public static void main(String[] args) {
ShareDataV3 v3 = new ShareDataV3();
new Thread(()->{
try {
v3.produce();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "AAA").start();
new Thread(()->{
try {
v3.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "BBB").start();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
v3.stop();
}
}
可以看到使用阻塞隊(duì)列根本不需要我們?nèi)ゼ渔i舅巷，通知什么的，完全解放了河咽。
運(yùn)行結(jié)果：
AAA 插入隊(duì)列1成功資源隊(duì)列大小= 0
BBB 消費(fèi)1成功 資源隊(duì)列大小= 0
AAA 插入隊(duì)列2成功資源隊(duì)列大小= 1
BBB 消費(fèi)2成功 資源隊(duì)列大小= 0
AAA 插入隊(duì)列3成功資源隊(duì)列大小= 1
BBB 消費(fèi)3成功 資源隊(duì)列大小= 1
AAA 插入隊(duì)列4成功資源隊(duì)列大小= 1
AAA 插入隊(duì)列5成功資源隊(duì)列大小= 2
BBB 消費(fèi)4成功 資源隊(duì)列大小= 1
AAA 插入隊(duì)列6成功資源隊(duì)列大小= 2
AAAFLAG變?yōu)閒lase钠右，生產(chǎn)停止
BBB 消費(fèi)5成功 資源隊(duì)列大小= 1
BBB 消費(fèi)6成功 資源隊(duì)列大小= 0
超過(guò)兩秒沒(méi)有取道數(shù)據(jù)，消費(fèi)者即將退出
Process finished with exit code 0
Java與大數(shù)據(jù)資源分享：
https://zhuanlan.zhihu.com/p/50030125
Java與大數(shù)據(jù)資源分享：
https://zhuanlan.zhihu.com/p/50030125
Java與大數(shù)據(jù)資源分享：
https://zhuanlan.zhihu.com/p/50030125
