生产者消费者_生产者消费者问题 - CSDN
精华内容
参与话题
  • 生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯,而通过阻塞队列来进行通讯,所以生产者生产完数据之后不用等待消费者处理,直接扔给阻塞队列,消费者不找...

    概念

    生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题生产者和消费者彼此之间不直接通讯,而通过阻塞队列来进行通讯,所以生产者生产完数据之后不用等待消费者处理,直接扔给阻塞队列,消费者不找生产者要数据,而是直接从阻塞队列里取,阻塞队列就相当于一个缓冲区,平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的

    321原则

    • 三种角色:生产者、消费者、仓库
    • 两种关系:生产者与生产者之间是互斥关系,消费者与消费者之间是互斥关系,生产者与消费者之间是同步与互斥关系。
    • 一个交易场所:仓库(这里我们用阻塞队列来表示)

    优点

    • 解耦–生产者。消费者之间不直接通信,降低了耦合度。
    • 支持并发
    • 支持忙闲不均
      在这里插入图片描述

    代码实现

    基于BlockingQueue的生产者消费者模型
    BlockingQueue 在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于,当队列为空时,从队列获取元素的操作将会被阻塞,直到队列中被放入了元素当队列满时,往队列里存放元素的操作也会被阻塞直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的,线程在对阻塞队列进程操作时会被阻塞)

      //这是生产者消费者模型的代码                                                                                                                   
      #include<iostream>
      #include<queue>
      #include<stdlib.h>
      #include<pthread.h>
      #include<unistd.h>
      
      #define NUM 8  
      
      class BlockQueue
      {
        private:
          std::queue<int> q;
          int cap;
          pthread_mutex_t mutex;
          pthread_cond_t full;
          pthread_cond_t empty;
      
        private:
          void LockQueue()  //队列加锁
          {
            pthread_mutex_lock(&mutex);
          }
          void UnlockQueue()  //队列解锁
          {
            pthread_mutex_unlock(&mutex);
          }
          void ProductWait()  //队列满,生产者等待
          {
            pthread_cond_wait(&full,&mutex);
          }
          void ConsumeWait()  //队列空,消费者等待
          {
            pthread_cond_wait(&empty,&mutex);
          }
          void NotifyProduct()  //队列不为满时,通知生产者
          {
            pthread_cond_signal(&full);
          }
          void NotifyConsume()  //队列不为空时,通知消费者
          {
            pthread_cond_signal(&empty);
          }
          bool IsEmpty()
          {
            return (q.size() == 0 ? true : false);
          }
          bool IsFull()
          {                                                                                                                                          
    	    return (q.size() == cap ? true : false);
          }
        public:
          BlockQueue(int _cap = NUM):cap(_cap) //构造函数
          {
            pthread_mutex_init(&mutex,NULL);
            pthread_cond_init(&full,NULL);
            pthread_cond_init(&empty,NULL);
          } 
          void PushData(const int &data)
          {
            LockQueue();
            while(IsFull()) //队列满
            {
              NotifyConsume();
              std::cout<<"queue full,notify consume data,product stop!!"<<std::endl;
              ProductWait();
            }
            //队列不满,生产者插入数据,通知消费者队列中已经有数据了
            q.push(data);
            NotifyConsume();
            UnlockQueue();
          }
          void PopData(int &data)
          {                                                                                                                                          
            LockQueue();
            while(IsEmpty())  //队列为空
            { 
              NotifyProduct();
              std::cout<<"queue empty,notify product data,consume stop!!"<<std::endl;
              ConsumeWait();
            }
            //队列不为空
            data = q.front();
            q.pop();
            NotifyProduct();
            UnlockQueue();
          }
          ~BlockQueue()
          {
            pthread_mutex_destroy(&mutex);
            pthread_cond_destroy(&full);
            pthread_cond_destroy(&empty);
          }
      };
      
      //消费者
      void* consumer(void* arg)
      {
        BlockQueue *bqp = (BlockQueue*)arg;
        int data;
        for(;;)                                                                                                                                      
        {
          bqp->PopData(data);
          std::cout<<"Consume data done: "<<data<<std::endl;
        }
      }
      
      //生产者
      void* producter(void* arg)
      {
        BlockQueue *bqp = (BlockQueue*)arg;
        srand((unsigned long)time(NULL));
        for(;;)
        {
          int data = rand()%1024;
          bqp->PushData(data);
          std::cout<<"Product data done: "<<data<<std::endl;
         // sleep(1);
        }
      }
      
      int main()
      {
        BlockQueue bq;
        pthread_t c,p;
      
        pthread_create(&c,NULL,consumer,(void*)&bq);
        pthread_create(&p,NULL,producter,(void*)&bq);
                                                                                                                                                     
        pthread_join(c,NULL);
        pthread_join(p,NULL);
      
        return 0;
      }
    
    

    运行结果:
    在这里插入图片描述

    展开全文
  • 【操作系统】生产者消费者问题

    万次阅读 多人点赞 2018-08-11 00:43:20
    生产者消费者模型 生产者消费者模型 一、 生产者消费者问题 二、 问题分析 三、 伪代码实现 四、代码实现(C++) 五、 互斥锁与条件变量的使用比较 一、 生产者消费者问题 生产者消费者问题...

    生产者消费者模型

    一、 生产者消费者问题

    生产者消费者问题(英语:Producer-consumer problem),也称有限缓冲问题(英语:Bounded-buffer problem),是一个多线程同步问题的经典案例。该问题描述了共享固定大小缓冲区的两个线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中,然后重复此过程。与此同时,消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据,消费者也不会在缓冲区中空时消耗数据。
    .
    要解决该问题,就必须让生产者在缓冲区满时休眠(要么干脆就放弃数据),等到下次消费者消耗缓冲区中的数据的时候,生产者才能被唤醒,开始往缓冲区添加数据。同样,也可以让消费者在缓冲区空时进入休眠,等到生产者往缓冲区添加数据之后,再唤醒消费者。通常采用进程间通信的方法解决该问题。如果解决方法不够完善,则容易出现死锁的情况。出现死锁时,两个线程都会陷入休眠,等待对方唤醒自己。该问题也能被推广到多个生产者和消费者的情形。

    这里写图片描述


    二、 问题分析

    该问题需要注意的几点:

    • 在缓冲区为空时,消费者不能再进行消费
    • 在缓冲区为满时,生产者不能再进行生产
    • 在一个线程进行生产或消费时,其余线程不能再进行生产或消费等操作,即保持线程间的同步
    • 注意条件变量与互斥锁的顺序

    这里写图片描述
    由于前两点原因,因此需要保持线程间的同步,即一个线程消费(或生产)完,其他线程才能进行竞争CPU,获得消费(或生产)的机会。对于这一点,可以使用条件变量进行线程间的同步:生产者线程在product之前,需要wait直至获取自己所需的信号量之后,才会进行product的操作;同样,对于消费者线程,在consume之前需要wait直到没有线程在访问共享区(缓冲区),再进行consume的操作,之后再解锁并唤醒其他可用阻塞线程。

    这里写图片描述
    在访问共享区资源时,为避免多个线程同时访问资源造成混乱,需要对共享资源加锁,从而保证某一时刻只有一个线程在访问共享资源。


    三、 伪代码实现

    假设缓冲区大小为10,生产者、消费者线程若干。生产者和消费者相互等效,只要缓冲池未满,生产者便可将消息送入缓冲池;只要缓冲池未空,消费者便可从缓冲池中取走一个消息。

    • items代表缓冲区已经使用的资源数,spaces代表缓冲区可用资源数
    • mutex代表互斥锁
    • buf[10] 代表缓冲区,其内容类型为item
    • in、out代表第一个资源和最后一个资源
    var items = 0, space = 10, mutex = 1;
    var in = 0, out = 0;
    item buf[10] = { NULL };
    
    producer {
        while( true ) {
            wait( space );  // 等待缓冲区有空闲位置, 在使用PV操作时,条件变量需要在互斥锁之前
            wait( mutex );  // 保证在product时不会有其他线程访问缓冲区
    
            // product
            buf.push( item, in );  // 将新资源放到buf[in]位置 
            in = ( in + 1 ) % 10;
            
            signal( mutex );  // 唤醒的顺序可以不同
            signal( items );  // 通知consumer缓冲区有资源可以取走
        }
    }
    
    consumer {
        while( true ) {
            wait( items );  // 等待缓冲区有资源可以使用
            wait( mutex );  // 保证在consume时不会有其他线程访问缓冲区
    
            // consume
            buf.pop( out );  // 将buf[out]位置的的资源取走
            out = ( out + 1 ) % 10;
    
            signal( mutex );  // 唤醒的顺序可以不同
            signal( space );  // 通知缓冲区有空闲位置
        }
    }
    

    不能将线程里两个wait的顺序调换否则会出现死锁。例如(调换后),将consumer的两个wait调换,在producer发出signal信号后,如果producer线程此时再次获得运行机会,执行完了wait(space),此时,另一个consumer线程获得运行机会,执行了 wait(mutex) ,如果此时缓冲区为空,那么consumer将会阻塞在wait(items),而producer也会因为无法获得锁的所有权所以阻塞在wait(mutex),这样两个线程都在阻塞,也就造成了死锁。


    四、代码实现(C++)

    #include <iostream>
    #include <string.h>
    #include <pthread.h>
    #include <unistd.h>
    using namespace std;
    
    int current = 0;  // producer运行加1,consumer运行减1
    int buf[10];
    int in = 0, out = 0;
    int items = 0, spaces = 10;
    bool flag;  // 标记线程结束运行
    pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
    pthread_cond_t notfull = PTHREAD_COND_INITIALIZER;  // 缓冲区不满
    pthread_cond_t notempty = PTHREAD_COND_INITIALIZER;  // 缓冲区不空
    
    void *producer( void *arg ) {
        while( flag ) {
            pthread_mutex_lock( &mutex );  // 为保证条件变量不会因为多线程混乱,所以先加锁
            while( !spaces ) {  // 避免“惊群”效应,避免因其他线程实现得到事件而导致该线程“假醒”
                pthread_cond_wait( &notfull, &mutex );
            }
            buf[in] = current++;
            in = ( in + 1 ) % 10;
            items++;
            spaces--;
    
            printf( "producer %zu , current = %d\n", pthread_self(), current );
            for( int i = 0; i < 10; i++ ) {
                printf( "%-4d", buf[i] );
            }
            printf( "\n\n" );
    
            pthread_cond_signal( &notempty );
            pthread_mutex_unlock( &mutex );
        }
        pthread_exit( NULL );
    }
    
    void *consumer( void *arg ) {
        while( flag ) {
            pthread_mutex_lock( &mutex );
            while( !items ) {
                pthread_cond_wait( &notempty, &mutex );
            }
            buf[out] = -1;
            out = ( out + 1 ) % 10;
            current--;
            items--;
            spaces++;
    
            printf( "consumer %zu , current = %d\n", pthread_self(), current );
            for( int i = 0; i < 10; i++ ) {
                printf( "%-4d", buf[i] );
            }
            printf( "\n\n" );
    
            pthread_cond_signal( &notfull );
            pthread_mutex_unlock( &mutex );
        }
        pthread_exit( NULL );
    }
    
    int main() {
        memset( buf, -1, sizeof(buf) );
        flag = true;
        pthread_t pro[10], con[10];
        int i = 0;
    
        for( int i = 0; i < 10; i++ ) {
            pthread_create( &pro[i], NULL, producer, NULL );
            pthread_create( &con[i], NULL, consumer, NULL );
        }
    
        sleep(1);  // 让线程运行一秒
        flag = false;
    
        for( int i = 0; i < 10; i++ ) {
            pthread_join( pro[i], NULL );
            pthread_join( con[i], NULL );
        }
    
        return 0;
    } 
    

    五、 互斥锁与条件变量的使用比较

    我们会发现,在伪代码中强调了条件变量在前,互斥锁在后,而到了代码实现时又变成了先加互斥锁,再进行循环pthread_cond_wait()。这不是自相矛盾吗?

    其实,在伪代码中的wait()signal()就是操作系统中的PV操作,而PV操作定义就保证了该语句是原子操作,因此在wait条件变量改变的时候不会因为多进程同时访问共享资源造成混乱,所以为了保证线程间的同步,需要先加条件变量,等事件可使用后才进行线程相应的操作,此时互斥锁的作用是保证共享资源不会被其他线程访问。

    而在代码实现中,signal()对应的时pthread_cond_wait()函数,该函数在执行时会有三步:

    • 解开当前的锁
    • 等待条件变量达到所需要的状态
    • 再把之前解开的锁加锁

    为了实现将pthread_cond_wait()变成原子操作,就需要在该函数之前添加互斥锁。因为pthread_cond_wait()可以解锁,也就不会发生像伪代码所说的死锁问题。相反,如果像伪代码那样先使用条件变量,后加锁,则会造成多个线程同时访问共享资源的问题,造成数据的混乱。


    欢迎关注微信公众号,不定时分享学习资料与学习笔记,感谢!
    在这里插入图片描述

    展开全文
  • JAVA多线程之生产者消费者模型

    千次阅读 2019-02-21 18:00:49
    生产者消费者模型 所谓的生产者消费者模型,是通过一个容器来解决生产者和消费者的强耦合问题。通俗的讲,就是生产者在不断的生产,消费者也在不断的消费,可是消费者消费的产品是生产者生产的,这就必然存在一个...

    生产者消费者模型

    所谓的生产者消费者模型,是通过一个容器来解决生产者和消费者的强耦合问题。通俗的讲,就是生产者在不断的生产,消费者也在不断的消费,可是消费者消费的产品是生产者生产的,这就必然存在一个中间容器,我们可以把这个容器想象成是一个货架,当货架空的时候,生产者要生产产品,此时消费者在等待生产者往货架上生产产品,而当货架满的时候,消费者可以从货架上拿走商品,生产者此时等待货架的空位,这样不断的循环。那么在这个过程中,生产者和消费者是不直接接触的,所谓的‘货架’其实就是一个阻塞队列,生产者生产的产品不直接给消费者消费,而是仍给阻塞队列,这个阻塞队列就是来解决生产者消费者的强耦合的。就是生产者消费者模型。

    总结一下:生产者消费者能够解决的问题如下:

    • 生产与消费的速度不匹配
    • 软件开发过程中解耦

    在具体实现生产者消费者模型之前需要先描述几个用到的方法:

    wait()

    先看一下wait()是干什么的?

    1.wait()是Object里面的方法,而不是Thread里面的,这一点很容易搞错。它的作用是将当前线程置于预执行队列,并在wait()所在的代码处停止,等待唤醒通知。
    2.wait()只能在同步代码块或者同步方法中执行,如果调用wait()方法,而没有持有适当的锁,就会抛出异常。
    wait()方法调用后悔释放出锁,线程与其他线程竞争重新获取锁。

    举个例子:

    public class TestWait implements Runnable {
        private final Object object=new Object();
        @Override
        public void run() {
            synchronized (object){
            System.out.println("线程执行开始。。。");
                try {
                    object.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("线程执行结束。。。");
            }
        }
    
        public static void main(String[] args) {
            TestWait testWait=new TestWait();
            Thread thread=new Thread(testWait);
            thread.start();
        }
    }
    

    结果如下:
    在这里插入图片描述

    从结果中我们可以看出线程调用了wait()方法后一直在等待,不会继续往下执行。这也就能解释上面说的wait()一旦执行,除非接收到唤醒操作或者是异常中断,否则不会继续往下执行。

    notify()方法

    在上面的代码中我们看到wait()调用以后线程一直在等待,在实际当中我们难免不希望是这样的,那么这个时候就用到了另一个方法notify方法:

    1.notify()方法也是要在同步代码块或者同步方法中调用的,它的作用是使停止的线程继续执行,调用notify()方法后,会通知那些等待当前线程对象锁的线程,并使它们重新获取该线程的对象锁,如果等待线程比较多的时候,则有线程规划器随机挑选出一个呈wait状态的线程。
    2.notify()调用之后不会立即释放锁,而是当执行notify()的线程执行完成,即退出同步代码块或同步方法时,才会释放对象锁。

    还是上面的例子,刚才我们调用了wait()方法后,线程便一直在等待,接下来我们给线程一个唤醒的信号,代码如下:

    public class TestWait implements Runnable {
        private final Object object=new Object();
    
        public void setFlag(boolean flag) {
            this.flag = flag;
        }
    
        private boolean flag=true;
        @Override
        public void run() {
            if(flag){
                this.testwait();
            }
            else {
                this.testnotify();
            }
    
        }
        public void testwait(){
            synchronized (object){
                try {
                System.out.println("线程开始执行。。。");
                    Thread.sleep(1000);
                object.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("线程执行结束。。。");
            }
        }
        public void testnotify(){
            synchronized (object){
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                object.notify();
            }
        }
    
        public static void main(String[] args) {
            TestWait testWait=new TestWait();
            Thread thread=new Thread(testWait);
            thread.start();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            testWait.setFlag(false);
            Thread thread1=new Thread(testWait);
            thread1.start();
        }
    }
    

    结果如下:
    在这里插入图片描述
    我们看到在调用notify()方法之后,线程又继续了。

    notifyAll()方法

    从字面意思就可以看出notifyAll是唤醒所有等待的线程。

    public class TestWait implements Runnable {
        private final Object object=new Object();
        private boolean flag=true;
        public void setFlag(boolean flag) {
            this.flag = flag;
        }
        @Override
        public void run() {
            if(flag){
                this.testwait();
            }
            else {
                this.testnotify();
            }
        }
        public void testwait(){
            synchronized (object){
                try {
                System.out.println(Thread.currentThread().getName()+"线程开始执行。。。");
                    Thread.sleep(1000);
                object.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName()+"线程执行结束。。。");
            }
        }
        public void testnotify(){
            synchronized (object){
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                object.notifyAll();
            }
        }
        public static void main(String[] args) {
            TestWait testWait=new TestWait();
            Thread thread=new Thread(testWait,"线程1");
            thread.start();
            Thread thread1=new Thread(testWait,"线程2");
            thread1.start();
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            testWait.setFlag(false);
            Thread thread2=new Thread(testWait);
            thread2.start();
    
        }
    }
    

    结果如下:
    在这里插入图片描述
    可见notifyAll()方法确实唤醒了所有等待的线程。

    小结

    出现阻塞的情况大体分为如下5种:

    1. 线程调用 sleep方法,主动放弃占用的处理器资源。
    2. 线程调用了阻塞式IO方法,在该方法返回前,该线程被阻塞。
    3. 线程试图获得一个同步监视器,但该同步监视器正被其他线程所持有。
    4. 线程等待某个通知。
    5. 程序调用了 suspend方法将该线程挂起。此方法容易导致死锁,尽量避免使用该方法。

    run()方法运行结束后进入销毁阶段,整个线程执行完毕。

    每个锁对象都有两个队列,一个是就绪队列,一个是阻塞队列。就绪队列存储了将要获得锁的线程,阻塞队列存储了被阻塞的线程。一个线程被唤醒后,才会进入就绪队列,等待CPU的调度;反之,一个线程被wait后,就会进入阻塞队列,等待下一次被唤醒。

    生产者消费者模型代码示例

    商品类

    public class Goods {
        private int id;
        private String name;
    
        public Goods(int id, String name) {
            this.id = id;
            this.name = name;
        }
    }
    

    生产者类

    public class Producer implements Runnable {
        private Goods goods;
    
        @Override
        public void run() {
                while (true) {
                    try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            synchronized (TestPC.queue) {
    
                    goods=new Goods(1,"商品");
                    if (TestPC.queue.size()<MAX_POOL) {
                        TestPC.queue.add(goods);
                        System.out.println(Thread.currentThread().getName()+"生产商品");
    
                    } else {
                        try {
                            TestPC.queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }
    }
    

    消费者类

    public class Consumer implements Runnable {
        @Override
        public void run() {
            while (true){
                try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized (TestPC.queue){
    
                if(!TestPC.queue.isEmpty()){
                    TestPC.queue.poll();
                    System.out.println(Thread.currentThread().getName()+"消费商品");
                }
                else {
                    TestPC.queue.notify();
                }
            }
        }
        }
    }
    

    测试类

    public class TestPC {
        public static final int MAX_POOL=10;
        public static final int MAX_PRODUCER=5;
        public static final int MAX_CONSUMER=4;
        public static  Queue<Goods> queue=new ArrayBlockingQueue<>(MAX_POOL);
        public static void main(String[] args) {
            Producer producer=new Producer();
            Consumer consumer=new Consumer();
            for(int i=0;i<MAX_PRODUCER;i++) {
                Thread threadA = new Thread(producer, "生产者线程"+i);
                threadA.start();
            }
            for(int j=0;j<MAX_CONSUMER;j++) {
                Thread threadB = new Thread(consumer, "消费者线程"+j);
                threadB.start();
            }
        }
    }
    
    

    部分结果展示:
    在这里插入图片描述

    展开全文
  • 生产者/消费者模式的理解及实现

    万次阅读 多人点赞 2018-05-31 10:05:37
    ★简介 生产者消费者模式并不是GOF提出的23种设计模式之一,23种设计模式都是建立在面向对象的基础之上的,但其实面向过程的编程中也有很多高效的编程模式,生产者消费者模式便是其中之一,它是我们编程过程中最...

      ★简介

       生产者消费者模式并不是GOF提出的23种设计模式之一,23种设计模式都是建立在面向对象的基础之上的,但其实面向过程的编程中也有很多高效的编程模式,生产者消费者模式便是其中之一,它是我们编程过程中最常用的一种设计模式。

       在实际的软件开发过程中,经常会碰到如下场景:某个模块负责产生数据,这些数据由另一个模块来负责处理(此处的模块是广义的,可以是类、函数、线程、进程等)。产生数据的模块,就形象地称为生产者;而处理数据的模块,就称为消费者。

        单单抽象出生产者和消费者,还够不上是生产者/消费者模式。该模式还需要有一个缓冲区处于生产者和消费者之间,作为一个中介。生产者把数据放入缓冲区,而消费者从缓冲区取出数据。大概的结构如下图。


       为了不至于太抽象,我们举一个寄信的例子(虽说这年头寄信已经不时兴,但这个例子还是比较贴切的)。假设你要寄一封平信,大致过程如下:

        1、你把信写好——相当于生产者制造数据

        2、你把信放入邮筒——相当于生产者把数据放入缓冲区

        3、邮递员把信从邮筒取出——相当于消费者把数据取出缓冲区

        4、邮递员把信拿去邮局做相应的处理——相当于消费者处理数据

        ★优点

       可能有同学会问了:这个缓冲区有什么用捏?为什么不让生产者直接调用消费者的某个函数,直接把数据传递过去?搞出这么一个缓冲区作甚?

        其实这里面是大有讲究的,大概有如下一些好处。

       ◇解耦

        假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法,那么生产者对于消费者就会产生依赖(也就是耦合)。将来如果消费者的代码发生变化,可能会影响到生产者。而如果两者都依赖于某个缓冲区,两者之间不直接依赖,耦合也就相应降低了。

        接着上述的例子,如果不使用邮筒(也就是缓冲区),你必须得把信直接交给邮递员。有同学会说,直接给邮递员不是挺简单的嘛?其实不简单,你必须得认识谁是邮递员,才能把信给他(光凭身上穿的制服,万一有人假冒,就惨了)。这就产生和你和邮递员之间的依赖(相当于生产者和消费者的强耦合)。万一哪天邮递员换人了,你还要重新认识一下(相当于消费者变化导致修改生产者代码)。而邮筒相对来说比较固定,你依赖它的成本就比较低(相当于和缓冲区之间的弱耦合)。

        ◇支持并发(concurrency)

        生产者直接调用消费者的某个方法,还有另一个弊端。由于函数调用是同步的(或者叫阻塞的),在消费者的方法没有返回之前,生产者只好一直等在那边。万一消费者处理数据很慢,生产者就会白白糟蹋大好时光。

        使用了生产者/消费者模式之后,生产者和消费者可以是两个独立的并发主体(常见并发类型有进程和线程两种,后面的帖子会讲两种并发类型下的应用)。生产者把制造出来的数据往缓冲区一丢,就可以再去生产下一个数据。基本上不用依赖消费者的处理速度。

        其实当初这个模式,主要就是用来处理并发问题的。

        从寄信的例子来看。如果没有邮筒,你得拿着信傻站在路口等邮递员过来收(相当于生产者阻塞);又或者邮递员得挨家挨户问,谁要寄信(相当于消费者轮询)。不管是哪种方法,都挺土的。

        ◇支持忙闲不均

        缓冲区还有另一个好处。如果制造数据的速度时快时慢,缓冲区的好处就体现出来了。当数据制造快的时候,消费者来不及处理,未处理的数据可以暂时存在缓冲区中。等生产者的制造速度慢下来,消费者再慢慢处理掉。

        为了充分复用,我们再拿寄信的例子来说事。假设邮递员一次只能带走1000封信。万一某次碰上情人节(也可能是圣诞节)送贺卡,需要寄出去的信超过1000封,这时候邮筒这个缓冲区就派上用场了。邮递员把来不及带走的信暂存在邮筒中,等下次过来时再拿走。

        费了这么多口水,希望原先不太了解生产者/消费者模式的同学能够明白它是怎么一回事。接下来说说数据单元。

        ★啥是数据单元

        何谓数据单元捏?简单地说,每次生产者放到缓冲区的,就是一个数据单元;每次消费者从缓冲区取出的,也是一个数据单元。对于前一个帖子中寄信的例子,我们可以把每一封单独的信件看成是一个数据单元。

        不过光这么介绍,太过于简单,无助于大伙儿分析出这玩意儿。所以,后面咱们来看一下数据单元需要具备哪些特性。搞明白这些特性之后,就容易从复杂的业务逻辑中分析出适合做数据单元的东西了。

        ★数据单元的特性

        分析数据单元,需要考虑如下几个方面的特性:

        ◇关联到业务对象

        首先,数据单元必须关联到某种业务对象。在考虑该问题的时候,你必须深刻理解当前这个生产者/消费者模式所对应的业务逻辑,才能够作出合适的判断。

        由于“寄信”这个业务逻辑比较简单,所以大伙儿很容易就可以判断出数据单元是啥。但现实生活中,往往没这么乐观。大多数业务逻辑都比较复杂,当中包含的业务对象是层次繁多、类型各异。在这种情况下,就不易作出决策了。

        这一步很重要,如果选错了业务对象,会导致后续程序设计和编码实现的复杂度大为上升,增加了开发和维护成本。

        ◇完整性

        所谓完整性,就是在传输过程中,要保证该数据单元的完整。要么整个数据单元被传递到消费者,要么完全没有传递到消费者。不允许出现部分传递的情形。

        对于寄信来说,你不能把半封信放入邮筒;同样的,邮递员从邮筒中拿信,也不能只拿出信的一部分。

        ◇独立性

        所谓独立性,就是各个数据单元之间没有互相依赖,某个数据单元传输失败不应该影响已经完成传输的单元;也不应该影响尚未传输的单元。

        为啥会出现传输失败捏?假如生产者的生产速度在一段时间内一直超过消费者的处理速度,那就会导致缓冲区不断增长并达到上限,之后的数据单元就会被丢弃。如果数据单元相互独立,等到生产者的速度降下来之后,后续的数据单元继续处理,不会受到牵连;反之,如果数据单元之间有某种耦合,导致被丢弃的数据单元会影响到后续其它单元的处理,那就会使程序逻辑变得非常复杂。

        对于寄信来说,某封信弄丢了,不会影响后续信件的送达;当然更不会影响已经送达的信件。

        ◇颗粒度

        前面提到,数据单元需要关联到某种业务对象。那么数据单元和业务对象是否要一一对应捏?很多场合确实是一一对应的。

        不过,有时出于性能等因素的考虑,也可能会把N个业务对象打包成一个数据单元。那么,这个N该如何取值就是颗粒度的考虑了。颗粒度的大小是有讲究的。太大的颗粒度可能会造成某种浪费;太小的颗粒度可能会造成性能问题。颗粒度的权衡要基于多方面的因素,以及一些经验值的考量。

        还是拿寄信的例子。如果颗粒度过小(比如设定为1),那邮递员每次只取出1封信。如果信件多了,那就得来回跑好多趟,浪费了时间。

        如果颗粒度太大(比如设定为100),那寄信的人得等到凑满100封信才拿去放入邮筒。假如平时很少写信,就得等上很久,也不太爽。

        可能有同学会问:生产者和消费者的颗粒度能否设置成不同大小(比如对于寄信人设置成1,对于邮递员设置成100)。当然,理论上可以这么干,但是在某些情况下会增加程序逻辑和代码实现的复杂度。后面讨论具体技术细节时,或许会聊到这个问题。

        好,数据单元的话题就说到这。希望通过本帖子,大伙儿能够搞明白数据单元到底是怎么一回事。下一个帖子,咱们来聊一下“基于队列的缓冲区”,技术上如何实现。

        [2]:队列缓冲区

        经过前面两个帖子的铺垫,今天终于开始聊一些具体的编程技术了。由于不同的缓冲区类型、不同的并发场景对于具体的技术实现有较大的影响。为了深入浅出、便于大伙儿理解,咱们先来介绍最传统、最常见的方式。也就是单个生产者对应单个消费者,当中用队列(FIFO)作缓冲。

        关于并发的场景,在之前的帖子“进程还线程?是一个问题!”中,已经专门论述了进程和线程各自的优缺点,两者皆不可偏废。所以,后面对各种缓冲区类型的介绍都会同时提及进程方式和线程方式。

        ★线程方式

        先来说一下并发线程中使用队列的例子,以及相关的优缺点。

        ◇内存分配的性能

        在线程方式下,生产者和消费者各自是一个线程。生产者把数据写入队列头(以下简称push),消费者从队列尾部读出数据(以下简称pop)。当队列为空,消费者就稍息(稍事休息);当队列满(达到最大长度),生产者就稍息。整个流程并不复杂。

        那么,上述过程会有什么问题捏?一个主要的问题是关于内存分配的性能开销。对于常见的队列实现:在每次push时,可能涉及到堆内存的分配;在每次pop时,可能涉及堆内存的释放。假如生产者和消费者都很勤快,频繁地push、pop,那内存分配的开销就很可观了。对于内存分配的开销,用Java的同学可以参见前几天的帖子“Java性能优化[1]”;对于用C/C++的同学,想必对OS底层机制会更清楚,应该知道分配堆内存(new或malloc)会有加锁的开销和用户态/核心态切换的开销。

        那该怎么办捏?请听下文分解,关于“生产者/消费者模式[3]:环形缓冲区”。

        ◇同步和互斥的性能

        另外,由于两个线程共用一个队列,自然就会涉及到线程间诸如同步啊、互斥啊、死锁啊等等劳心费神的事情。好在"操作系统"这门课程对此有详细介绍,学过的同学应该还有点印象吧?对于没学过这门课的同学,也不必难过,网上相关的介绍挺多的(比如"这里"),大伙自己去瞅一瞅。关于这方面的细节,咱今天就不多啰嗦了。

        这会儿要细谈的是,同步和互斥的性能开销。在很多场合中,诸如信号量、互斥量等玩意儿的使用也是有不小的开销的(某些情况下,也可能导致用户态/核心态切换)。如果像刚才所说,生产者和消费者都很勤快,那这些开销也不容小觑啊。

        这又该咋办捏?请听下文的下文分解,关于“生产者/消费者模式[4]:双缓冲区”。

        ◇适用于队列的场合

        刚才尽批判了队列的缺点,难道队列方式就一无是处?非也。由于队列是很常见的数据结构,大部分编程语言都内置了队列的支持(具体介绍见"这里"),有些语言甚至提供了线程安全的队列(比如JDK 1.5引入的ArrayBlockingQueue)。因此,开发人员可以捡现成,避免了重新发明轮子。

        所以,假如你的数据流量不是很大,采用队列缓冲区的好处还是很明显的:逻辑清晰、代码简单、维护方便。比较符合KISS原则。

        ★进程方式

        说完了线程的方式,再来介绍基于进程的并发。

        跨进程的生产者/消费者模式,非常依赖于具体的进程间通讯(IPC)方式。而IPC的种类名目繁多,不便于挨个列举(毕竟口水有限)。因此咱们挑选几种跨平台、且编程语言支持较多的IPC方式来说事儿。

        ◇匿名管道

        感觉管道是最像队列的IPC类型。生产者进程在管道的写端放入数据;消费者进程在管道的读端取出数据。整个的效果和线程中使用队列非常类似,区别在于使用管道就无需操心线程安全、内存分配等琐事(操作系统暗中都帮你搞定了)。

        管道又分命名管道和匿名管道两种,今天主要聊匿名管道。因为命名管道在不同的操作系统下差异较大(比如Win32和POSIX,在命名管道的API接口和功能实现上都有较大差异;有些平台不支持命名管道,比如Windows CE)。除了操作系统的问题,对于有些编程语言(比如Java)来说,命名管道是无法使用的。所以我一般不推荐使用这玩意儿。

        其实匿名管道在不同平台上的API接口,也是有差异的(比如Win32的CreatePipe和POSIX的pipe,用法就很不一样)。但是我们可以仅使用标准输入和标准输出(以下简称stdio)来进行数据的流入流出。然后利用shell的管道符把生产者进程和消费者进程关联起来(没听说过这种手法的同学,可以看"这里")。实际上,很多操作系统(尤其是POSIX风格的)自带的命令都充分利用了这个特性来实现数据的传输(比如more、grep等)。

        这么干有几个好处:

        1、基本上所有操作系统都支持在shell方式下使用管道符。因此很容易实现跨平台。

        2、大部分编程语言都能够操作stdio,因此跨编程语言也就容易实现。

        3、刚才已经提到,管道方式省却了线程安全方面的琐事。有利于降低开发、调试成本。

        当然,这种方式也有自身的缺点:

        1、生产者进程和消费者进程必须得在同一台主机上,无法跨机器通讯。这个缺点比较明显。

        2、在一对一的情况下,这种方式挺合用。但如果要扩展到一对多或者多对一,那就有点棘手了。所以这种方式的扩展性要打个折扣。假如今后要考虑类似的扩展,这个缺点就比较明显。

        3、由于管道是shell创建的,对于两边的进程不可见(程序看到的只是stdio)。在某些情况下,导致程序不便于对管道进行操纵(比如调整管道缓冲区尺寸)。这个缺点不太明显。

        4、最后,这种方式只能单向传数据。好在大多数情况下,消费者进程不需要传数据给生产者进程。万一你确实需要信息反馈(从消费者到生产者),那就费劲了。可能得考虑换种IPC方式。

        顺便补充几个注意事项,大伙儿留意一下:

        1、对stdio进行读写操作是以阻塞方式进行。比如管道中没有数据,消费者进程的读操作就会一直停在哪儿,直到管道中重新有数据。

        2、由于stdio内部带有自己的缓冲区(这缓冲区和管道缓冲区是两码事),有时会导致一些不太爽的现象(比如生产者进程输出了数据,但消费者进程没有立即读到)。具体的细节,大伙儿可以看"这里"。

        ◇SOCKET(TCP方式)

        基于TCP方式的SOCKET通讯是又一个类似于队列的IPC方式。它同样保证了数据的顺序到达;同样有缓冲的机制。而且这玩意儿也是跨平台和跨语言的,和刚才介绍的shell管道符方式类似。

        SOCKET相比shell管道符的方式,有啥优点捏?主要有如下几个优点:

        1、SOCKET方式可以跨机器(便于实现分布式)。这是主要优点。

        2、SOCKET方式便于将来扩展成为多对一或者一对多。这也是主要优点。

        3、SOCKET可以设置阻塞和非阻塞方法,用起来比较灵活。这是次要优点。

        4、SOCKET支持双向通讯,有利于消费者反馈信息。

        当然有利就有弊。相对于上述shell管道的方式,使用SOCKET在编程上会更复杂一些。好在前人已经做了大量的工作,搞出很多SOCKET通讯库和框架给大伙儿用(比如C++的ACE库、Python的Twisted)。借助于这些第三方的库和框架,SOCKET方式用起来还是比较爽的。由于具体的网络通讯库该怎么用不是本系列的重点,此处就不细说了。

        虽然TCP在很多方面比UDP可靠,但鉴于跨机器通讯先天的不可预料性(比如网线可能被某傻X给拔错了,网络的忙闲波动可能很大),在程序设计上我们还是要多留一手。具体该如何做捏?可以在生产者进程和消费者进程内部各自再引入基于线程的"生产者/消费者模式"。这话听着像绕口令,为了便于理解,画张图给大伙儿瞅一瞅。


    这么做的关键点在于把代码分为两部分:生产线程和消费线程属于和业务逻辑相关的代码(和通讯逻辑无关);发送线程和接收线程属于通讯相关的代码(和业务逻辑无关)。

        这样的好处是很明显的,具体如下:

        1、能够应对暂时性的网络故障。并且在网络故障解除后,能够继续工作。

        2、网络故障的应对处理方式(比如断开后的尝试重连),只影响发送和接收线程,不会影响生产线程和消费线程(业务逻辑部分)。

        3、具体的SOCKET方式(阻塞和非阻塞)只影响发送和接收线程,不影响生产线程和消费线程(业务逻辑部分)。

        4、不依赖TCP自身的发送缓冲区和接收缓冲区。(默认的TCP缓冲区的大小可能无法满足实际要求)

        5、业务逻辑的变化(比如业务需求变更)不影响发送线程和接收线程。

        针对上述的最后一条,再多啰嗦几句。如果整个业务系统中有多个进程是采用上述的模式,那或许可以重构一把:在业务逻辑代码和通讯逻辑代码之间切一刀,把业务逻辑无关的部分封装成一个通讯中间件(说中间件显得比较牛X :-)。如果大伙儿对这玩意儿有兴趣,以后专门开个帖子聊。

        [3]:环形缓冲区

        前一个帖子提及了队列缓冲区可能存在的性能问题及解决方法:环形缓冲区。今天就专门来描述一下这个话题。

        为了防止有人给咱扣上“过度设计”的大帽子,事先声明一下:只有当存储空间的分配/释放非常频繁并且确实产生了明显的影响,你才应该考虑环形缓冲区的使用。否则的话,还是老老实实用最基本、最简单的队列缓冲区吧。还有一点需要说明一下:本文所提及的“存储空间”,不仅包括内存,还可能包括诸如硬盘之类的存储介质。

        ★环形缓冲区 vs 队列缓冲区

        ◇外部接口相似

        在介绍环形缓冲区之前,咱们先来回顾一下普通的队列。普通的队列有一个写入端和一个读出端。队列为空的时候,读出端无法读取数据;当队列满(达到最大尺寸)时,写入端无法写入数据。

        对于使用者来讲,环形缓冲区和队列缓冲区是一样的。它也有一个写入端(用于push)和一个读出端(用于pop),也有缓冲区“满”和“空”的状态。所以,从队列缓冲区切换到环形缓冲区,对于使用者来说能比较平滑地过渡。

        ◇内部结构迥异

        虽然两者的对外接口差不多,但是内部结构和运作机制有很大差别。队列的内部结构此处就不多啰嗦了。重点介绍一下环形缓冲区的内部结构。

        大伙儿可以把环形缓冲区的读出端(以下简称R)和写入端(以下简称W)想象成是两个人在体育场跑道上追逐(R追W)。当R追上W的时候,就是缓冲区为空;当W追上R的时候(W比R多跑一圈),就是缓冲区满。

        为了形象起见,去找来一张图并略作修改,如下:


        从上图可以看出,环形缓冲区所有的push和pop操作都是在一个固定的存储空间内进行。而队列缓冲区在push的时候,可能会分配存储空间用于存储新元素;在pop时,可能会释放废弃元素的存储空间。所以环形方式相比队列方式,少掉了对于缓冲区元素所用存储空间的分配、释放。这是环形缓冲区的一个主要优势。

        ★环形缓冲区的实现

        如果你手头已经有现成的环形缓冲区可供使用,并且你对环形缓冲区的内部实现不感兴趣,可以跳过这段。

        ◇数组方式 vs 链表方式

        环形缓冲区的内部实现,即可基于数组(此处的数组,泛指连续存储空间)实现,也可基于链表实现。

        数组在物理存储上是一维的连续线性结构,可以在初始化时,把存储空间一次性分配好,这是数组方式的优点。但是要使用数组来模拟环,你必须在逻辑上把数组的头和尾相连。在顺序遍历数组时,对尾部元素(最后一个元素)要作一下特殊处理。访问尾部元素的下一个元素时,要重新回到头部元素(第0个元素)。如下图所示:


        使用链表的方式,正好和数组相反:链表省去了头尾相连的特殊处理。但是链表在初始化的时候比较繁琐,而且在有些场合(比如后面提到的跨进程的IPC)不太方便使用。

        ◇读写操作

        环形缓冲区要维护两个索引,分别对应写入端(W)和读取端(R)。写入(push)的时候,先确保环没满,然后把数据复制到W所对应的元素,最后W指向下一个元素;读取(pop)的时候,先确保环没空,然后返回R对应的元素,最后R指向下一个元素。

        ◇判断“空”和“满”

        上述的操作并不复杂,不过有一个小小的麻烦:空环和满环的时候,R和W都指向同一个位置!这样就无法判断到底是“空”还是“满”。大体上有两种方法可以解决该问题。

        办法1:始终保持一个元素不用

        当空环的时候,R和W重叠。当W比R跑得快,追到距离R还有一个元素间隔的时候,就认为环已经满。当环内元素占用的存储空间较大的时候,这种办法显得很土(浪费空间)。

        办法2:维护额外变量

        如果不喜欢上述办法,还可以采用额外的变量来解决。比如可以用一个整数记录当前环中已经保存的元素个数(该整数>=0)。当R和W重叠的时候,通过该变量就可以知道是“空”还是“满”。

        ◇元素的存储

        由于环形缓冲区本身就是要降低存储空间分配的开销,因此缓冲区中元素的类型要选好。尽量存储值类型的数据,而不要存储指针(引用)类型的数据。因为指针类型的数据又会引起存储空间(比如堆内存)的分配和释放,使得环形缓冲区的效果打折扣。

        ★应用场合

        刚才介绍了环形缓冲区内部的实现机制。按照前一个帖子的惯例,我们来介绍一下在线程和进程方式下的使用。

        如果你所使用的编程语言和开发库中带有现成的、成熟的环形缓冲区,强烈建议使用现成的库,不要重新制造轮子;确实找不到现成的,才考虑自己实现。如果你纯粹是业余时间练练手,那另当别论。

        ◇用于并发线程

        和线程中的队列缓冲区类似,线程中的环形缓冲区也要考虑线程安全的问题。除非你使用的环形缓冲区的库已经帮你实现了线程安全,否则你还是得自己动手搞定。线程方式下的环形缓冲区用得比较多,相关的网上资料也多,下面就大致介绍几个。

        对于C++的程序员,强烈推荐使用boost提供的circular_buffer模板,该模板最开始是在boost 1.35版本中引入的。鉴于boost在C++社区中的地位,大伙儿应该可以放心使用该模板。

        对于C程序员,可以去看看开源项目circbuf,不过该项目是GPL协议的,不太爽;而且活跃度不太高;而且只有一个开发人员。大伙儿慎用!建议只拿它当参考。

        对于C#程序员,可以参考CodeProject上的一个示例。

        ◇用于并发进程

        进程间的环形缓冲区,似乎少有现成的库可用。大伙儿只好自己动手、丰衣足食了。

        适用于进程间环形缓冲的IPC类型,常见的有共享内存和文件。在这两种方式上进行环形缓冲,通常都采用数组的方式实现。程序事先分配好一个固定长度的存储空间,然后具体的读写操作、判断“空”和“满”、元素存储等细节就可参照前面所说的来进行。

        共享内存方式的性能很好,适用于数据流量很大的场景。但是有些语言(比如Java)对于共享内存不支持。因此,该方式在多语言协同开发的系统中,会有一定的局限性。

        而文件方式在编程语言方面支持很好,几乎所有编程语言都支持操作文件。但它可能会受限于磁盘读写(Disk I/O)的性能。所以文件方式不太适合于快速数据传输;但是对于某些“数据单元”很大的场合,文件方式是值得考虑的。

        对于进程间的环形缓冲区,同样要考虑好进程间的同步、互斥等问题,限于篇幅,此处就不细说了。



    生产/消费者问题是个非常典型的多线程问题,涉及到的对象包括“生产者”、“消费者”、“仓库”和“产品”。他们之间的关系如下:

    ① 生产者仅仅在仓储未满时候生产,仓满则停止生产。

    ② 消费者仅仅在仓储有产品时候才能消费,仓空则等待。

    ③ 当消费者发现仓库没产品可消费时候会通知生产者生产。

    ④ 生产者在生产出可消费产品时候,应该通知等待的消费者去消费。

    wait/notify/notifyAll实现和用LockCondition实现。

    用wait/notify/notifyAll 实现生产者消费者模型:

    方法一:用五个类来实现,分别为Produce(实现生产过程), Consumer(实现消费过程), ProduceThread(实现生产者线程),ConsumeThread(实现消费者线程),Main等。需要注意的是有两个地方。

    ① 用while判断当前list是否为空;

    ② 调用的是object的notifyAll()方法而不是notify()方法。

    方法二:用四个类实现,分别为MyService(实现生产和消费过程用synchronized关键字实现同步),ProduceThread(实现生产者线程),ConsumeThread(实现消费者线程),Main。需要注意的也是方法一中的两个地方while和notifyAll()。

    LockCondition实现。共有四个类,分别是分别为MyService(实现生产和消费过程,用lock实现线程间同步),ProduceThread(实现生产者线程),ConsumeThread(实现消费者线程),Main。需要注意的也是方法一中的两个地方while和signalAll()。


    方法一:

    [java] view plain copy
    1. package ProduceConsumer;  
    2.   
    3. import java.util.ArrayList;  
    4.   
    5. public class Produce {  
    6.       
    7.     public Object object;  
    8.     public ArrayList<Integer> list;//用list存放生产之后的数据,最大容量为1  
    9.               
    10.     public Produce(Object object,ArrayList<Integer> list ){  
    11.         this.object = object;  
    12.         this.list = list;  
    13.     }  
    14.       
    15.     public void produce() {  
    16.           
    17.         synchronized (object) {  
    18.             /*只有list为空时才会去进行生产操作*/  
    19.             try {  
    20.             while(!list.isEmpty()){  
    21.                     System.out.println("生产者"+Thread.currentThread().getName()+" waiting");  
    22.                     object.wait();  
    23.                 }   
    24.             int value = 9999;  
    25.             list.add(value);  
    26.             System.out.println("生产者"+Thread.currentThread().getName()+" Runnable");  
    27.             object.notifyAll();//然后去唤醒因object调用wait方法处于阻塞状态的线程  
    28.         }catch (InterruptedException e) {  
    29.                 e.printStackTrace();  
    30.             }  
    31.         }  
    32.     }  
    33.   
    34. }  

    [java] view plain copy
    1. package ProduceConsumer;  
    2.   
    3. import java.util.ArrayList;  
    4.   
    5. public class Consumer {  
    6.       
    7.     public Object object;  
    8.     public ArrayList<Integer> list;//用list存放生产之后的数据,最大容量为1  
    9.               
    10.     public Consumer(Object object,ArrayList<Integer> list ){  
    11.         this.object = object;  
    12.         this.list = list;  
    13.     }  
    14.       
    15.     public void consmer() {  
    16.           
    17.         synchronized (object) {  
    18.             try {  
    19.                 /*只有list不为空时才会去进行消费操作*/  
    20.                 while(list.isEmpty()){  
    21.                     System.out.println("消费者"+Thread.currentThread().getName()+" waiting");  
    22.                     object.wait();  
    23.                 }   
    24.             list.clear();  
    25.             System.out.println("消费者"+Thread.currentThread().getName()+" Runnable");  
    26.             object.notifyAll();//然后去唤醒因object调用wait方法处于阻塞状态的线程  
    27.               
    28.         }catch (InterruptedException e) {  
    29.             e.printStackTrace();  
    30.         }  
    31.     }  
    32.     }  
    33.       
    34. }  

    [java] view plain copy
    1. package ProduceConsumer;  
    2.   
    3. public class ProduceThread extends Thread {  
    4.     private Produce p;  
    5.     public ProduceThread(Produce p){  
    6.         this.p = p;  
    7.     }  
    8.     @Override  
    9.     public void run() {  
    10.         while (true) {  
    11.             p.produce();  
    12.         }  
    13.     }  
    14. }  

    [java] view plain copy
    1. package ProduceConsumer;  
    2.   
    3. public class ConsumeThread extends Thread {  
    4.     private Consumer c;  
    5.     public ConsumeThread(Consumer c){  
    6.         this.c = c;  
    7.     }  
    8.     @Override  
    9.     public void run() {  
    10.         while (true) {  
    11.             c.consmer();  
    12.         }  
    13.     }  
    14. }  

    [java] view plain copy
    1. package ProduceConsumer;  
    2.   
    3. import java.util.ArrayList;  
    4.   
    5. public class Main {  
    6.     public static void main(String[] args) {  
    7.         Object object = new Object();  
    8.         ArrayList<Integer> list = new ArrayList<Integer>();  
    9.       
    10.         Produce p = new Produce(object, list);  
    11.         Consumer c = new Consumer(object, list);  
    12.           
    13.         ProduceThread[] pt = new ProduceThread[2];  
    14.         ConsumeThread[] ct = new ConsumeThread[2];  
    15.           
    16.         for(int i=0;i<2;i++){  
    17.             pt[i] = new ProduceThread(p);  
    18.             pt[i].setName("生产者 "+(i+1));  
    19.             ct[i] = new ConsumeThread(c);  
    20.             ct[i].setName("消费者"+(i+1));  
    21.             pt[i].start();  
    22.             ct[i].start();  
    23.         }  
    24.     }  
    25. }  

    方法二:

    [java] view plain copy
    1. package ProduceConsumer2;  
    2.   
    3. import java.util.ArrayList;  
    4.   
    5. public class MyService {  
    6.       
    7.     public ArrayList<Integer> list = new ArrayList<Integer>();//用list存放生产之后的数据,最大容量为1  
    8.     synchronized public void produce() {  
    9.               
    10.         try {  
    11.             /*只有list为空时才会去进行生产操作*/  
    12.             while(!list.isEmpty()){  
    13.                     System.out.println("生产者"+Thread.currentThread().getName()+" waiting");  
    14.                     this.wait();  
    15.                 }   
    16.             int value = 9999;  
    17.             list.add(value);  
    18.             System.out.println("生产者"+Thread.currentThread().getName()+" Runnable");  
    19.             this.notifyAll();//然后去唤醒因object调用wait方法处于阻塞状态的线程  
    20.         }catch (InterruptedException e) {  
    21.                 e.printStackTrace();  
    22.             }  
    23.     }  
    24.       
    25.     synchronized public void consmer() {  
    26.         try {     
    27.             /*只有list不为空时才会去进行消费操作*/  
    28.             while(list.isEmpty()){  
    29.                     System.out.println("消费者"+Thread.currentThread().getName()+" waiting");  
    30.                     this.wait();  
    31.             }  
    32.             list.clear();  
    33.             System.out.println("消费者"+Thread.currentThread().getName()+" Runnable");  
    34.             this.notifyAll();//然后去唤醒因object调用wait方法处于阻塞状态的线程  
    35.         } catch (InterruptedException e) {  
    36.             e.printStackTrace();  
    37.         }  
    38.     }  
    39.       
    40. }  

    [java] view plain copy
    1. package ProduceConsumer2;  
    2.   
    3. public class ProduceThread extends Thread {  
    4.     private MyService p;  
    5.     public ProduceThread(MyService p){  
    6.         this.p = p;  
    7.     }  
    8.     @Override  
    9.     public void run() {  
    10.         while (true) {  
    11.             p.produce();  
    12.         }  
    13.     }  
    14. }  

    [java] view plain copy
    1. package ProduceConsumer2;  
    2.   
    3. public class ConsumeThread extends Thread {  
    4.     private MyService c;  
    5.     public ConsumeThread(MyService c){  
    6.         this.c = c;  
    7.     }  
    8.     @Override  
    9.     public void run() {  
    10.         while (true) {  
    11.             c.consmer();  
    12.         }  
    13.     }  
    14. }  

    LockCondition实现

    [java] view plain copy
    1. package ConditionProduceConsumer;  
    2.   
    3. import java.util.concurrent.locks.Condition;  
    4. import java.util.concurrent.locks.ReentrantLock;  
    5.   
    6. public class MyService {  
    7.       
    8.     private ReentrantLock lock = new ReentrantLock();  
    9.     private Condition condition = lock.newCondition();  
    10.     private boolean hasValue = false;  
    11.       
    12.       
    13.      public void produce() {  
    14.         lock.lock();  
    15.         try {  
    16.             /*只有list为空时才会去进行生产操作*/  
    17.             while(hasValue == true){  
    18.                 System.out.println("生产者"+Thread.currentThread().getName()+" waiting");  
    19.                 condition.await();  
    20.             }  
    21.             hasValue = true;  
    22.             System.out.println("生产者"+Thread.currentThread().getName()+" Runnable");  
    23.             condition.signalAll();//然后去唤醒因object调用wait方法处于阻塞状态的线程  
    24.         } catch (InterruptedException e) {  
    25.             e.printStackTrace();  
    26.         }finally{  
    27.             lock.unlock();  
    28.         }  
    29.               
    30.     }  
    31.       
    32.      public void consmer() {  
    33.         lock.lock();  
    34.         try {  
    35.             /*只有list为空时才会去进行生产操作*/  
    36.             while(hasValue == false){  
    37.                 System.out.println("消费者"+Thread.currentThread().getName()+" waiting");  
    38.                 condition.await();  
    39.             }  
    40.             hasValue = false;  
    41.             System.out.println("消费者"+Thread.currentThread().getName()+" Runnable");  
    42.             condition.signalAll();//然后去唤醒因object调用wait方法处于阻塞状态的线程  
    43.         } catch (InterruptedException e) {  
    44.             e.printStackTrace();  
    45.         }finally{  
    46.             lock.unlock();  
    47.         }  
    48.       
    49.     }  
    50. }  

    [java] view plain copy
    1. package ConditionProduceConsumer;  
    2.   
    3. public class ProduceThread extends Thread {  
    4.     private MyService p;  
    5.     public ProduceThread(MyService p){  
    6.         this.p = p;  
    7.     }  
    8.     @Override  
    9.     public void run() {  
    10.         while (true) {  
    11.             p.produce();  
    12.         }  
    13.     }  
    14. }  
    [java] view plain copy
    1. package ConditionProduceConsumer;  
    2.   
    3. public class ConsumeThread extends Thread {  
    4.     private MyService c;  
    5.     public ConsumeThread(MyService c){  
    6.         this.c = c;  
    7.     }  
    8.     @Override  
    9.     public void run() {  
    10.         while (true) {  
    11.             c.consmer();  
    12.         }  
    13.     }  
    14. }  

    [java] view plain copy
    1. package ConditionProduceConsumer;  
    2.   
    3.   
    4. public class Main {  
    5.     public static void main(String[] args) {  
    6.   
    7.         MyService service = new MyService();  
    8.           
    9.         ProduceThread[] pt = new ProduceThread[2];  
    10.         ConsumeThread[] ct = new ConsumeThread[2];  
    11.           
    12.         for(int i=0;i<1;i++){  
    13.             pt[i] = new ProduceThread(service);  
    14.             pt[i].setName("Condition 生产者 "+(i+1));  
    15.             ct[i] = new ConsumeThread(service);  
    16.             ct[i].setName("Condition 消费者"+(i+1));  
    17.             pt[i].start();  
    18.             ct[i].start();  
    19.         }  
    20.     }  
    21. }  

    展开全文
  • Java多种方式解决生产者消费者问题(十分详细)

    万次阅读 多人点赞 2018-08-16 08:40:50
    生产者消费者问题 一、问题描述 生产者消费者问题(Producer-consumer problem),也称有限缓冲问题(Bounded-buffer problem),是一个多线程同步问题的经典案例。生产者生成一定量的数据放到缓冲区中,...
  • 生产者消费者模型java实现

    千次阅读 2018-09-12 10:33:45
    做题的时候遇到了生产者消费者问题,这个问题可以说是线程学习的经典题目了,就忍不住研究了一波。它描述是有一块缓冲区(队列实现)作为仓库,生产者可以将产品放入仓库,消费者则可以从仓库中取走产品。在Java中这...
  •  生产者消费者模式并不是GOF提出的23种设计模式之一,23种设计模式都是建立在面向对象的基础之上的,但其实面向过程的编程中也有很多高效的编程模式,生产者消费者模式便是其中之一,它是我们编程过程中最常用的一...
  • 秒杀多线程第十篇 生产者消费者问题

    万次阅读 多人点赞 2012-05-21 10:18:09
    继经典线程同步问题之后,我们来看看生产者消费者问题及读者写者问题。生产者消费者问题是一个著名的线程同步问题,该问题描述如下:有一个生产者在生产产品,这些产品将提供给若干个消费者去消费,为了使生产者和...
  • 信号量与生产者消费者问题

    万次阅读 多人点赞 2017-01-19 15:06:44
    生产者—消费者问题 生产者—消费者题型在各类考试(考研、程序员证书、程序员面试笔试、期末考试)很常见,原因之一是生产者...生产者—消费者题型最基本的是有界缓冲区的生产者消费者问题和无界缓冲区的生产者消费者
  • 生产者消费者模型你知道多少

    万次阅读 多人点赞 2013-11-12 08:07:10
    进入正题之前先说点故事。从最开始学java的那里开始:我是从08年下半年开始学...比较清楚的记得马士兵讲到生产者消费者模型的时候还大谈特谈要是掌握了这个工资可以+1000(现在回忆起有点像历史一样,多少有些伤感)。
  • C++ 生产者消费者模式的简单实现

    千次阅读 2019-08-20 11:47:03
    生产者消费者模式可以理解为在生产者和消费者之间添加一个缓冲区,生产者只负责向缓冲区添加元素,而消费者只负责从缓冲区提取元素并使用。 这么做可以对生产者与消费者进行解耦,这样一来消费者不直接调用生产者,...
  • 生产者消费者问题是线程模型中的经典问题:生产者和消费者在同一时间段内共用同一存储空间,生产者向空间里生产数据,而消费者取走数据。 阻塞队列就相当于一个缓冲区,平衡了生产者和消费者的处理能力。这个阻塞...
  • Linux实现生产者消费者模型

    千次阅读 2017-06-01 11:35:03
    生产者消费者模型 简单来说就是“321原则(并非某一规则,而是为了理解生产者消费者模型)” “3”代表的是三种关系 生产者与消费者的互斥与同步关系 生产者与生产者的互斥(或竞争)关系 消费者与消费者的...
  • 实现生产者消费者的三种方式

    千次阅读 多人点赞 2019-10-14 21:22:03
    文章目录wait/notify的消息通知机制预备知识wait/notify消息通知潜在的一些问题notify过早通知等待wait的条件发生变化假死状态wait/notifyAll实现生产者-消费者使用Lock中Condition的await/signalAll实现生产者-消费...
  • 阅读本篇之前,如果你还没有看过从生产者消费者窥探线程同步(上) ,那不妨先戳一下,两篇一起嚼才更好呢。上一篇分析了使用BlockQueue和synchronized来实现生产者消费者模式。这一篇来看一下其他的实现,闲言少叙...
  • 操作系统------生产者消费者模型

    千次阅读 2018-04-24 18:06:59
    在“进程间通信----信号量”一文中,有简单介绍过生产者消费者模型的基本概念。在下文中将使用有关线程的互斥与同步的相关概念来实现两种不同类型的生产者消费者模型。在本文中侧重于线程间同步的实现。有关线程互斥...
  • 关于生产者消费者模式的C#实现

    千次阅读 2018-12-25 09:12:39
    记录一下用C#实现生产者消费者模式吧。 先介绍一下这个模式,简而言之就是生产者(可能有数个)生产东西,消费者(可能有数个)消费前面生产的东西。举个生活中的例子就是苹果有好几个厂家(生产者)生产iphone,线...
  • Java线程实现生产者消费者模式

    千次阅读 2018-06-11 20:24:46
    1 什么是生产者消费者模式想一个现实生活中的例子,啤酒商---超市---消费者也就是我们,啤酒商生产了啤酒,然后将啤酒销售给了超市,我们消费之又会到超市将啤酒买回来自己喝,那么啤酒商和消费者之间是什么关系呢?...
  • 生产者消费者问题 C++实现

    万次阅读 2018-06-29 17:00:03
    生产者消费者问题 C++实现 知识准备 thread 介绍 成员类 成员函数 sleep_for 介绍 mutex 介绍 成员函数 unique_lock 介绍 成员函数 codition_variable 介绍 成员函数 代码示例 生产者消费者问题 ...
  • 生产者消费者模式+代码实现

    千次阅读 2018-04-04 16:29:25
    在并发编程中使用生产者消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。为什么要使用生产者消费者模式在线程世界里,生产者就是生产数据的...
1 2 3 4 5 ... 20
收藏数 252,648
精华内容 101,059
关键字:

生产者消费者