生产者和消费者模型

生产者和消费者模型

1. 什么是生产者和消费者模型

生产者消费者模型具体来讲,就是在一个系统中,存在生产者和消费者两种角色,他们通过内存缓冲区进行通信,生产者生产消费者需要的资料,消费者把资料做成产品。

再具体一点:

  1. 生产者生产数据到缓冲区中,消费者从缓冲区中取数据。
  2. 如果缓冲区已经满了,则生产者线程阻塞。
  3. 如果缓冲区为空,那么消费者线程阻塞。

2. 如何实现

实现生产者消费者模型有两种方式:

  1. 采用 wait—notify 方式实现生产者消费者模型(注意这里需要加同步锁 synchronized)
  2. 采用 阻塞队列 方式实现生产者消费者模式

3. wait-notify 方式

实现过程并不复杂,直接上代码:

这里设置了生产者生产速度大于消费者消费速度(通过 sleep() 方法实现)。

缓冲区 BufferArea.java

public class BufferArea {

    // 当前资源数量的计数值
    private int currNum = 0;

    // 资源池中允许存放的资源数目
    private int maxSize = 10;

    /**
     * 从资源池中取走资源
     */
    public synchronized void get() {
        if (currNum > 0) {
            currNum--;
            System.out.println("Cosumer_" + Thread.currentThread().getName() + "消耗一件资源," + "当前缓冲区有" + currNum + "个");
            // 通知生产者生产资源
            notifyAll();
        } else {
            try {
                // 如果没有资源,则 Cosumer_ 进入等待状态
                System.out.println("Cosumer_" + Thread.currentThread().getName() + ": 当前缓冲区资源不足,进入等待状态");
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 向缓冲区中添加资源
     */
    public synchronized void put() {
        // 若当前缓冲区内的资源计数小于最大 size 数,才加
        if (currNum < maxSize) {
            currNum++;
            System.out.println(Thread.currentThread().getName() + "生产一件资源,当前资源池有" + currNum + "个");

            // 通知等待的消费者
            notifyAll();
        } else {
            // 若当前缓冲区的资源计数大于最大 size 数,则等待
            try {
                System.out.println(Thread.currentThread().getName() + "线程进入等待 << 当前缓冲区的资源计数大于最大 size 数");
                // 生产者进入等待状态,并释放锁
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

生产者 Producer.java

public class Producer extends Thread {

    private BlockQueueBufferArea mBufferArea;

    public Producer(BlockQueueBufferArea bufferArea) {
        this.mBufferArea = bufferArea;
        setName("Producer_" + getName());
    }

    @Override
    public void run() {
        // 不断的生产资源
        while (true) {
            sleepSomeTime();
            mBufferArea.put();
        }
    }

    private void sleepSomeTime() {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

消费者 Consumer

public class Consumer extends Thread {

    private BlockQueueBufferArea mBufferArea;

    public Consumer(BlockQueueBufferArea bufferArea) {
        this.mBufferArea = bufferArea;
        setName("Consumer_" + getName());
    }

    @Override
    public void run() {
        // 不断的取出资源
        while (true) {
            sleepSomeTime();
            mBufferArea.get();
        }
    }

    private void sleepSomeTime() {
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

测试 Test.java

public class Test {

    public static void main(String[] args) {
        BlockQueueBufferArea bufferArea = new BlockQueueBufferArea();

        Consumer consumer1 = new Consumer(bufferArea);
        Consumer consumer2 = new Consumer(bufferArea);
        Consumer consumer3 = new Consumer(bufferArea);

        Producer producer1 = new Producer(bufferArea);
        Producer producer2 = new Producer(bufferArea);
        Producer producer3 = new Producer(bufferArea);

        consumer1.start();
        consumer2.start();
        consumer3.start();

        producer1.start();
        producer2.start();
        producer3.start();

    }

}

打印结果如下:

ProducerThread-5生产一件资源,当前资源池有1个
ProducerThread-4生产一件资源,当前资源池有2个
ProducerThread-3生产一件资源,当前资源池有3个
ProducerThread-5生产一件资源,当前资源池有4个
ProducerThread-4生产一件资源,当前资源池有5个
ProducerThread-3生产一件资源,当前资源池有6个
ProducerThread-5生产一件资源,当前资源池有7个
ProducerThread-4生产一件资源,当前资源池有8个
ProducerThread-3生产一件资源,当前资源池有9个
ProducerThread-3生产一件资源,当前资源池有10个
ProducerThread-4线程进入等待 << 当前缓冲区的资源计数大于最大 size 数
ProducerThread-5线程进入等待 << 当前缓冲区的资源计数大于最大 size 数
ProducerThread-3线程进入等待 << 当前缓冲区的资源计数大于最大 size 数

>> 注释:3个生产者线程生产满了10个(maxSize)产品,然后就都进入了等待

Cosumer_Consumer_Thread-0消耗一件资源,当前缓冲区有9个
Cosumer_Consumer_Thread-1消耗一件资源,当前缓冲区有8个
Cosumer_Consumer_Thread-2消耗一件资源,当前缓冲区有7个

>> 注释:3个消费者消费了3个产品

ProducerThread-3生产一件资源,当前资源池有8个
ProducerThread-5生产一件资源,当前资源池有9个
ProducerThread-4生产一件资源,当前资源池有10个

>> 注释:生产者立马又生产3个

...

>> 然后一直循环往复这个过程

4. 阻塞队列方式

阻塞队列的特点:
  • 当队列元素已满的时候,阻塞插入操作
  • 当队列元素为空的时候,阻塞获取操作
不同的阻塞队列:

ArrayBlockingQueue 与 LinkedBlockingQueue 都是支持 FIFO (先进先出),但是 LinkedBlockingQueue 是无界的,而ArrayBlockingQueue 是有界的。

这里我们采用无界阻塞队列来演示生产者消费者模式。

演示

还是设置生产者生产速度大于消费者消费速度(通过 sleep() 方法实现)

缓冲区 BlockQueueBufferArea.java

public class BlockQueueBufferArea {

    BlockingQueue<Integer> mProductPoll = new LinkedBlockingQueue(10);

    public void  put() {
        try {
            System.out.println(Thread.currentThread().getName() + "产品池被放入了一个资源");
            mProductPoll.put(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void get() {
        try {
            System.out.println(Thread.currentThread().getName() + "产品池被取走了一个资源");
            mProductPoll.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

生产者 Producer.java

public class Producer extends Thread {

    private BlockQueueBufferArea mBufferArea;

    public Producer(BlockQueueBufferArea bufferArea) {
        this.mBufferArea = bufferArea;
        setName("Producer_" + getName());
    }

    @Override
    public void run() {
        // 不断的生产资源
        while (true) {
            sleepSomeTime();
            mBufferArea.put();
        }
    }

    private void sleepSomeTime() {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

消费者 Consumer.java

public class Consumer extends Thread {

    private BlockQueueBufferArea mBufferArea;

    public Consumer(BlockQueueBufferArea bufferArea) {
        this.mBufferArea = bufferArea;
        setName("Consumer_" + getName());
    }

    @Override
    public void run() {
        // 不断的取出资源
        while (true) {
            sleepSomeTime();
            mBufferArea.get();
        }
    }

    private void sleepSomeTime() {
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

测试 Test.java

public class Test {

    public static void main(String[] args) {
        BlockQueueBufferArea bufferArea = new BlockQueueBufferArea();

        Consumer consumer1 = new Consumer(bufferArea);
        Consumer consumer2 = new Consumer(bufferArea);
        Consumer consumer3 = new Consumer(bufferArea);

        Producer producer1 = new Producer(bufferArea);
        Producer producer2 = new Producer(bufferArea);
        Producer producer3 = new Producer(bufferArea);

        consumer1.start();
        consumer2.start();
        consumer3.start();

        producer1.start();
        producer2.start();
        producer3.start();

    }

}

打印结果如下:

Producer_Thread-5产品池被放入了一个资源
Producer_Thread-4产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源
Producer_Thread-4产品池被放入了一个资源
Producer_Thread-5产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源
Producer_Thread-4产品池被放入了一个资源
Producer_Thread-5产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源
Producer_Thread-4产品池被放入了一个资源
Producer_Thread-5产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源
Consumer_Thread-0产品池被取走了一个资源
Consumer_Thread-1产品池被取走了一个资源
Consumer_Thread-2产品池被取走了一个资源
Producer_Thread-4产品池被放入了一个资源
Producer_Thread-5产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源

5. 参考

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

推荐阅读更多精彩内容