Carson带你学Java:解决生产者、消费者问题的五种实现方式


1. 简介

生产者 & 消费者之间存在 强耦合问题


2. 解决方案

采用 生产者 & 消费者 模式,具体介绍如下:

示意图

3. 具体解决方式介绍

方式1:wait() / notify()

// Object类里的两个方法,所有Object子类都可使用这2个方法
// 对象的监视器对锁对象的锁定(也就是代码中的lock对象),注意是调用锁对象的wait() / nofity()
public class Test { 
    private static Integer count = 0; 
    private final Integer FULL = 5; 
    private static String lock = "lock"; 
 
    public static void main(String[] args) { 
        Test t = new Test(); 
        new Thread(t.new Producer()).start(); 
        new Thread(t.new Consumer()).start(); 
        new Thread(t.new Producer()).start(); 
        new Thread(t.new Consumer()).start(); 
    } 
 
    class Producer implements Runnable { 
        @Override 
        public void run() { 
            for (int i = 0; i < 5; i++) { 
                try { 
                    Thread.sleep(1000); 
                } catch (InterruptedException e1) {  
                    e1.printStackTrace(); 
                } 
                synchronized (lock) { 
                    while (count == FULL) { 
                        try { 
                            lock.wait(); 
                        } catch (InterruptedException e) {  
                            e.printStackTrace(); 
                        } 
                    } 
                    count++; 
                    System.out.println("生产者"+Thread.currentThread().getName() 
                            + "已生产完成,商品数量:" + count); 
                    lock.notifyAll(); 
                } 
            } 
        } 
    } 
 
    class Consumer implements Runnable { 
        @Override 
        public void run() { 
            for (int i = 0; i < 5; i++) { 
                try { 
                    Thread.sleep(1000); 
                } catch (InterruptedException e1) { 
                    e1.printStackTrace(); 
                } 
                synchronized (lock) { 
                    while (count == 0) { 
                        try { 
                            lock.wait(); 
                        } catch (InterruptedException e) { 
                            e.printStackTrace(); 
                        } 
                    } 
                    count--; 
                    System.out.println("消费者"+Thread.currentThread().getName() 
                            + "已消费,剩余商品数量:" + count); 
                    lock.notifyAll(); 
                } 
            } 
        } 
    } 
}


// 测试结果

生产者Thread-0已生产完成,商品数量:1 
生产者Thread-2已生产完成,商品数量:2 
消费者Thread-1已消费,剩余商品数量:1 
消费者Thread-3已消费,剩余商品数量:0 
生产者Thread-0已生产完成,商品数量:1 
消费者Thread-3已消费,剩余商品数量:0 
生产者Thread-2已生产完成,商品数量:1 
消费者Thread-1已消费,剩余商品数量:0 
生产者Thread-0已生产完成,商品数量:1 
生产者Thread-2已生产完成,商品数量:2 
消费者Thread-1已消费,剩余商品数量:1 
消费者Thread-3已消费,剩余商品数量:0 
生产者Thread-0已生产完成,商品数量:1 
消费者Thread-1已消费,剩余商品数量:0 
生产者Thread-2已生产完成,商品数量:1 
消费者Thread-3已消费,剩余商品数量:0 
生产者Thread-0已生产完成,商品数量:1 
消费者Thread-1已消费,剩余商品数量:0 
生产者Thread-2已生产完成,商品数量:1 
消费者Thread-3已消费,剩余商品数量:0

方式2:await() / signal()

// 对wait()/notify()的改进,功能更加强大、更适用于高级用户
// synchronized 托管给JVM执行
// 而lock是Java写的控制锁的代码

import java.util.concurrent.locks.Condition; 
import java.util.concurrent.locks.Lock; 
import java.util.concurrent.locks.ReentrantLock; 
 
public class Test { 
    private static Integer count = 0;// 缓冲区 
    private final Integer FULL = 5; 
    final Lock lock = new ReentrantLock(); // 此处采用 ReentrantLock,获得可重入锁 
    final Condition put = lock.newCondition(); 
    final Condition get = lock.newCondition(); 
 
    public static void main(String[] args) { 
 
        Test t = new Test(); 
        new Thread(t.new Producer()).start(); 
        new Thread(t.new Consumer()).start(); 
        new Thread(t.new Consumer()).start(); 
        new Thread(t.new Producer()).start(); 
    } 
 
    // 生产者 
    class Producer implements Runnable { 
        @Override 
        public void run() { 
            for (int i = 0; i < 5; i++) { 
                try { 
                    Thread.sleep(1000); 
                } catch (InterruptedException e1) {  
                    e1.printStackTrace(); 
                } 
                //加锁 
                lock.lock(); 
                try { 
                    while (count == FULL) { 
                        try { 
                            put.await(); 
                        } catch (InterruptedException e) {  
                            e.printStackTrace(); 
                        } 
                    } 
                    count++; 
                    System.out.println("生产者" + Thread.currentThread().getName() 
                            + "已生产完成,商品数量: " + count); 
                    //通知消费者,现在可以消费 
                    get.signal(); 
                } finally { 
                    lock.unlock(); 
                } 
            } 
        } 
    } 
 
    // 消费者 
    class Consumer implements Runnable { 
 
        @Override 
        public void run() { 
            for (int i = 0; i < 5; i++) { 
                try { 
                    Thread.sleep(1000); 
                } catch (InterruptedException e1) {  
                    e1.printStackTrace(); 
                } 
                lock.lock(); 
                try { 
                    while (count == 0) { 
                        try { 
                            get.await(); 
                        } catch (Exception e) {  
                            e.printStackTrace(); 
                        } 
                    } 
                    count--; 
                    System.out.println("消费者" + Thread.currentThread().getName() 
                            + "已消费,剩余商品数量: " + count); 
                    put.signal(); 
                } finally { 
                    lock.unlock(); 
                } 
            } 
        } 
    } 
}

// 测试结果
生产者Thread-3已生产完成,商品数量: 1 
生产者Thread-0已生产完成,商品数量: 2 
消费者Thread-1已消费,剩余商品数量: 1 
消费者Thread-2已消费,剩余商品数量: 0 
生产者Thread-3已生产完成,商品数量: 1 
生产者Thread-0已生产完成,商品数量: 2 
消费者Thread-1已消费,剩余商品数量: 1 
消费者Thread-2已消费,剩余商品数量: 0 
生产者Thread-0已生产完成,商品数量: 1 
生产者Thread-3已生产完成,商品数量: 2 
消费者Thread-1已消费,剩余商品数量: 1 
消费者Thread-2已消费,剩余商品数量: 0 
生产者Thread-0已生产完成,商品数量: 1 
生产者Thread-3已生产完成,商品数量: 2 
消费者Thread-2已消费,剩余商品数量: 1 
消费者Thread-1已消费,剩余商品数量: 0 
生产者Thread-3已生产完成,商品数量: 1 
生产者Thread-0已生产完成,商品数量: 2 
消费者Thread-2已消费,剩余商品数量: 1 
消费者Thread-1已消费,剩余商品数量: 0

方式3:(BlockingQueue)阻塞队列 系列方法

示意图
// 下面主要使用其中的 put()、take()
// put():将指定元素插入此队列中,将等待可用的空间(若有必要)
// take():获取并移除此队列的头部,在指定的等待时间前等待可用的元素(若有必要)

import java.util.concurrent.ArrayBlockingQueue; 
import java.util.concurrent.BlockingQueue; 
 
public class Test { 
    private static Integer count = 0; 
    final BlockingQueue<Integer> bq = new ArrayBlockingQueue<Integer>(5);// 容量为5的阻塞队列 
 
    public static void main(String[] args) { 
        Test t = new Test(); 
        new Thread(t.new Producer()).start(); 
        new Thread(t.new Consumer()).start(); 
        new Thread(t.new Consumer()).start(); 
        new Thread(t.new Producer()).start(); 
    } 
 
// 生产者
    class Producer implements Runnable { 
        @Override 
        public void run() { 
            for (int i = 0; i < 5; i++) { 
                try { 
                    Thread.sleep(1000); 
                } catch (Exception e) { 
                    e.printStackTrace(); 
                } 
                try { 
                    bq.put(1); 
                    count++; 
                    System.out.println("生产者" + Thread.currentThread().getName() 
                            + "已生产完成,商品数量:" + count); 
                } catch (InterruptedException e) { 
                    e.printStackTrace(); 
                } 
            } 
        } 
    } 
 
// 消费者
    class Consumer implements Runnable { 
 
        @Override 
        public void run() { 
            for (int i = 0; i < 5; i++) { 
                try { 
                    Thread.sleep(1000); 
                } catch (InterruptedException e1) { 
                    e1.printStackTrace(); 
                } 
                try { 
                    bq.take(); 
                    count--; 
                    System.out.println("消费者" + Thread.currentThread().getName() 
                            + "已消费,剩余商品数量:" + count); 
                } catch (Exception e) { 
                    e.printStackTrace(); 
                } 
            } 
        } 
    } 
}

方式4:信号量 系列方法(Semaphore)

示意图
// 实例使用
import java.util.concurrent.Semaphore; 
 
public class Test { 
    int count = 0; 
    final Semaphore put = new Semaphore(5);// 初始令牌个数 
    // 注:同步令牌(notFull.acquire())必须在互斥令牌(mutex.acquire())前面获得;若先得到互斥锁再发生等待,会造成死锁。
    final Semaphore get = new Semaphore(0); 
    final Semaphore mutex = new Semaphore(1); 
 
    public static void main(String[] args) { 
        Test t = new Test(); 
        new Thread(t.new Producer()).start(); 
        new Thread(t.new Consumer()).start(); 
        new Thread(t.new Consumer()).start(); 
        new Thread(t.new Producer()).start(); 
    } 
 
    class Producer implements Runnable { 
        @Override 
        public void run() { 
            for (int i = 0; i < 5; i++) { 
                try { 
                    Thread.sleep(1000); 
                } catch (Exception e) { 
                    e.printStackTrace(); 
                } 
                try { 
                    put.acquire();// 注意顺序 
                    mutex.acquire(); 
                    count++; 
                    System.out.println("生产者" + Thread.currentThread().getName() 
                            + "已生产完成,商品数量:" + count); 
                } catch (Exception e) { 
                    e.printStackTrace(); 
                } finally { 
                    mutex.release(); 
                    get.release(); 
                } 
 
            } 
        } 
    } 
 
    class Consumer implements Runnable { 
 
        @Override 
        public void run() { 
            for (int i = 0; i < 5; i++) { 
                try { 
                    Thread.sleep(1000); 
                } catch (InterruptedException e1) { 
                    e1.printStackTrace(); 
                } 
                try { 
                    get.acquire();// 注意顺序 
                    mutex.acquire(); 
                    count--; 
                    System.out.println("消费者" + Thread.currentThread().getName() 
                            + "已消费,剩余商品数量:" + count); 
                } catch (Exception e) { 
                    e.printStackTrace(); 
                } finally { 
                    mutex.release(); 
                    put.release(); 
                } 
            } 
        } 
    } 
}

// 测试结果
生产者Thread-0已生产完成,商品数量:1 
消费者Thread-2已消费,剩余商品数量:0 
生产者Thread-3已生产完成,商品数量:1 
消费者Thread-1已消费,剩余商品数量:0 
生产者Thread-0已生产完成,商品数量:1 
生产者Thread-3已生产完成,商品数量:2 
消费者Thread-2已消费,剩余商品数量:1 
消费者Thread-1已消费,剩余商品数量:0 
生产者Thread-0已生产完成,商品数量:1 
生产者Thread-3已生产完成,商品数量:2 
消费者Thread-2已消费,剩余商品数量:1 
消费者Thread-1已消费,剩余商品数量:0 
生产者Thread-0已生产完成,商品数量:1 
生产者Thread-3已生产完成,商品数量:2 
消费者Thread-2已消费,剩余商品数量:1 
消费者Thread-1已消费,剩余商品数量:0 
生产者Thread-0已生产完成,商品数量:1 
生产者Thread-3已生产完成,商品数量:2 
消费者Thread-2已消费,剩余商品数量:1 
消费者Thread-1已消费,剩余商品数量:0

方式5:PipedInputStream / PipedOutputStream

示意图
import java.io.IOException; 
import java.io.PipedInputStream; 
import java.io.PipedOutputStream; 
 
public class Test { 
    final PipedInputStream pis = new PipedInputStream(); 
    final PipedOutputStream pos = new PipedOutputStream(); 
 
    public static void main(String[] args) { 
        Test t = new Test(); 
        new Thread(t.new Producer()).start(); 
        new Thread(t.new Consumer()).start(); 
    } 
 
    class Producer implements Runnable { 
 
        @Override 
        public void run() { 
            try { 
                pis.connect(pos); 
            } catch (IOException e) { 
                e.printStackTrace(); 
            } 
            try { 
                while (true) { // 不断的产生数据 
                    int n = (int) (Math.random() * 255); 
                    System.out.println("生产者" + Thread.currentThread().getName() 
                            + "已生产完成,商品数量:" + n); 
                    pos.write(n); 
                    pos.flush(); 
                } 
            } catch (IOException e) { 
                e.printStackTrace(); 
            } finally { 
                try { 
                    pis.close(); 
                    pos.close(); 
                } catch (IOException e) { 
                    e.printStackTrace(); 
                } 
            } 
 
        } 
    } 
 
    class Consumer implements Runnable { 
 
        @Override 
        public void run() { 
            int n; 
            try { 
                while (true) { 
                    n = pis.read(); 
                    System.out.println("消费者" + Thread.currentThread().getName() 
                            + "已消费,剩余商品数量:" + n); 
                } 
            } catch (IOException e) { 
                e.printStackTrace(); 
            } finally { 
                try { 
                    pis.close(); 
                    pos.close(); 
                } catch (IOException e) { 
                    e.printStackTrace(); 
                } 
 
            } 
        } 
    } 
}

// 测试结果
生产者Thread-0已生产完成,商品数量:6 
生产者Thread-0已生产完成,商品数量:158 
生产者Thread-0已生产完成,商品数量:79 
生产者Thread-0已生产完成,商品数量:119 
生产者Thread-0已生产完成,商品数量:93 
生产者Thread-0已生产完成,商品数量:213 
生产者Thread-0已生产完成,商品数量:151 
生产者Thread-0已生产完成,商品数量:101 
生产者Thread-0已生产完成,商品数量:125 
生产者Thread-0已生产完成,商品数量:109 
生产者Thread-0已生产完成,商品数量:67 
生产者Thread-0已生产完成,商品数量:109 
生产者Thread-0已生产完成,商品数量:132 
生产者Thread-0已生产完成,商品数量:139 
... 



至此,关于Java解决生产者、消费者问题的五种实现方式讲解完毕。


欢迎关注Carson_Ho的简书!

分享Android技术干货,追求短、平、快,但却不缺深度


请点赞!因为你的鼓励是我写作的最大动力!

相关文章阅读
Android开发:最全面、最易懂的Android屏幕适配解决方案
Android事件分发机制详解:史上最全面、最易懂
Android开发:史上最全的Android消息推送解决方案
Android开发:最全面、最易懂的Webview详解
Android开发:JSON简介及最全面解析方法!
Android四大组件:Service服务史上最全面解析
Android四大组件:BroadcastReceiver史上最全面解析

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,098评论 5 476
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,213评论 2 380
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 149,960评论 0 336
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,519评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,512评论 5 364
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,533评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,914评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,574评论 0 256
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,804评论 1 296
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,563评论 2 319
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,644评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,350评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,933评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,908评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,146评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,847评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,361评论 2 342

推荐阅读更多精彩内容