/**
* 線程并發(fā)協(xié)作/線程通信
* 多線程并發(fā)協(xié)作模型“生產(chǎn)者/消費者模式”
* 模式分為三部分，生產(chǎn)者模塊/線程+緩沖區(qū)+消費者模塊/線程
* 通過緩沖區(qū)將生產(chǎn)者與消費者之間的耦合解除，生產(chǎn)者產(chǎn)出數(shù)據(jù)交給緩沖區(qū)，消費者從緩沖區(qū)提取數(shù)據(jù)，這也解決了忙閑不均的問題
*/
public class TestThread3 {
}
class Data1{
? ?//數(shù)據(jù)
? ?private byte[] data;

? ?public Data1(byte[] data) {
? ? ? ?this.data = data;
? ?}

? ?public byte[] getData() {
? ? ? ?return data;
? ?}

? ?public void setData(byte[] data) {
? ? ? ?this.data = data;
? ?}
}
class Buffer1{
? ?//緩沖區(qū)
? ?public final Data1[] buff = new Data1[10];
? ?//最大緩存10份數(shù)據(jù)
? ?public int size;
? ?//緩存中的數(shù)據(jù)量

? ?public boolean isFulled(){
? ? ? ?return size==10;
? ?}

? ?public synchronized void push(Data1 d){
? ? ? ?//存數(shù)據(jù)，存取數(shù)據(jù)時synchronized(this)串行
? ? ? ?while (isFulled()){
? ? ? ? ? ?//當緩存已滿時讓生產(chǎn)者線程進入阻塞，通過循環(huán)使每次從阻塞變?yōu)榫途w變?yōu)檫\行時重新判斷緩存狀況，直至不滿結束循環(huán)存入數(shù)據(jù)
? ? ? ? ? ?try {
? ? ? ? ? ? ? ?this.wait();
? ? ? ? ? ? ? ?//.wait()方法屬于Object類，只能在synchronized語句塊中使用，將當前對象鎖釋放并同時將當前線程變?yōu)樽枞麪顟B(tài)，.wait()的阻塞需要.notify()喚醒
? ? ? ? ? ? ? ?//this.wait()將this釋放
? ? ? ? ? ?} catch (InterruptedException e) {
? ? ? ? ? ? ? ?throw new RuntimeException(e);
? ? ? ? ? ?}
? ? ? ?}
? ? ? ?buff[size++]=d;
? ? ? ?System.out.println(Thread.currentThread().getName()+" 存入第"+size+"個數(shù)據(jù)");
? ? ? ?this.notify();
? ? ? ?//Object類的.notify()，喚醒一個this對象鎖的阻塞線程
? ?}

? ?public boolean isEmpty(){
? ? ? ?return size==0;
? ?}

? ?public Data1 pop(){
? ? ? ?//取數(shù)據(jù)
? ? ? ?Data1 d;
? ? ? ?synchronized (this) {
? ? ? ? ? ?while (isEmpty()) {
? ? ? ? ? ? ? ?try {
? ? ? ? ? ? ? ? ? ?this.wait();
? ? ? ? ? ? ? ? ? ?//緩存為空的時候阻塞
? ? ? ? ? ? ? ? ? ?//.wait(long ms)等待指定毫秒后自動喚醒
? ? ? ? ? ? ? ? ? ?//.wait(long ms,int ns)等待多少毫秒多少微秒后自動喚醒
? ? ? ? ? ? ? ?} catch (InterruptedException e) {
? ? ? ? ? ? ? ? ? ?throw new RuntimeException(e);
? ? ? ? ? ? ? ?}
? ? ? ? ? ?}
? ? ? ? ? ?System.out.println(Thread.currentThread().getName() + " 取出第" + size + "個數(shù)據(jù)");
? ? ? ? ? ?d = buff[--size];
? ? ? ? ? ?buff[size]=null;
? ? ? ? ? ?this.notify();
? ? ? ? ? ?//.notifyAll()喚醒同一對象鎖上所有調(diào)用.wait()方法的線程，優(yōu)先級高的線程先運行
? ? ? ?}
? ? ? ?return d;
? ?}
}

class Maker1 extends Thread{
? ?//生產(chǎn)者，沒有生產(chǎn)產(chǎn)品之前，消費者要進入等待狀態(tài)，而生產(chǎn)產(chǎn)品之后，又需要馬上通知消費者消費
? ?Buffer1 buff;

? ?public Maker1(Buffer1 buff,String name) {
? ? ? ?super(name);
? ? ? ?this.buff = buff;
? ?}

? ?@Override
? ?public void run() {
? ? ? ?for (int i=0;i<10;i++){
? ? ? ? ? ?buff.push(new Data1(("數(shù)據(jù)"+i).getBytes()));
? ? ? ?}
? ?}
}
class Consumer1 extends Thread{
? ?//消費者，在消費之后，要通知生產(chǎn)者已經(jīng)消費結束，需要繼續(xù)生產(chǎn)新產(chǎn)品
? ?Buffer1 buff;

? ?public Consumer1(Buffer1 buff,String name) {
? ? ? ?super(name);
? ? ? ?this.buff = buff;
? ?}

? ?@Override
? ?public void run() {
? ? ? ?for (int i=0;i<20;i++){
? ? ? ? ? ?buff.pop();
? ? ? ?}
? ?}
}
class Test3{
? ?public static void main(String[] args) {
? ? ? ?Buffer1 buff = new Buffer1();
? ? ? ?new Maker1(buff,"生產(chǎn)者1").start();
? ? ? ?new Maker1(buff,"生產(chǎn)者2").start();
? ? ? ?new Consumer1(buff,"消費者1").start();
? ?}
}