Java并发之生产者和消费者

?生产者和消费者问题是操作系统并发性经典问题之一,描述的是有一块缓冲区为仓库,生产者可以将商品放入仓库,消费者可以从仓库拿走商品,解决生产者和消费者问题的方法可以分为两类:1.采用某种机制保护生产者和消费者之间的同步。2.在生产者和消费者之间建立一个管道。
?第2种方法不常用,暂不做讨论。同步的核心问题是如何保证同一资源在多线程并发访问时的完整性。常用的同步的方法是采用信号或加锁机制,保证资源在任意时刻之多被一个线程访问。在Java中一共有四种方法支撑同步:

  1. wait/notify
  2. await/signal
  3. BlockingQueue
  4. PipedInputStream/PipedOutStream

前三种是同步方法,最后一种是管道方法,本文先容前三种。

1.wait/notify(notifyAll)

wait,notify,notifyAll是基类Object的三个方法,它们必须是由同步监视器(synchronized包围的那个对象)来调用,用于协调多个线程对共享数据的存储。
wait:使当前线程暂停实行并释放对象锁标志,让其它线程可以进入synchronized数据快,当前线程被放入对象等待池中。
notify:从对象等待池中取走任意一个线程放入锁标志等待池中,如果锁标志等待池中没有线程,则notify不起作用。
notifyAll:从对象等待池中取走所有的线程放入对象等待池中。
注意:notify从对象等待线程池中取出的线程有可能不是希望的线程,而notifyAll有可能取出不必要的线程,影响性能。
下面贴上实现代码。
首先定义一个资源:

public class PublicResource {
   private int number = 0;
   private int size = 10;

   /**
    * 生产
    */
   public synchronized int increase() {

       while (number >= size) {
           System.out.println(Thread.currentThread().getName() + "资源已满,生产者等待...");
           try {
               //对象的wait方法被调用后,线程进入对象的等待队列中,并释放对象锁,其它线程可以竞争使用此对象锁,
               //sleep方法使得一个线程进入睡眠状态,但是线程所占有的资源并没有释放。
               wait();
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
       ++number;
       System.out.println(Thread.currentThread().getName() + "---生产者---生产了1个,总共有" + number);
       notifyAll();
       return number;
   }

   /**
    * 消费
    */
   public synchronized int decrease() {
       while (number <= 0) {
           System.out.println(Thread.currentThread().getName() + "---资源已空,消费者等待...");
           try {
               wait();
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
       --number;
       System.out.println(Thread.currentThread().getName() + "---消费者---消费了1个,总共有" + number);
       notifyAll();
       return number;
   }

   /**
    * 获取资源大小
    *
    * @return 资源大小
    */
   public int getSize() {
       return size;
   }

}

贴上生产者代码:

public class IncreaseRunnable implements Runnable {
    private PublicResource mPublicResource;

    public IncreaseRunnable(PublicResource publicResource) {
        mPublicResource = publicResource;
    }

    @Override
    public void run() {

        //生产一个资源
        mPublicResource.increase();

    }

}

贴上消费者代码:

public class DecreaseRunnable implements Runnable {
    private PublicResource mPublicResource;

    public DecreaseRunnable(PublicResource publicResource) {
        mPublicResource = publicResource;
    }

    @Override
    public void run() {
        mPublicResource.decrease();
    }
}

最后贴上main方法代码:

public class ProcuderConsumerDemo {

    public static void main(String[] arg0) {
        PublicResource publicResource = new PublicResource();
        new Thread(new IncreaseRunnable(publicResource)).start();//生产一个资源线程
        new Thread(new DecreaseRunnable(publicResource)).start();//消费一个资源线程

        new Thread(new IncreaseRunnable(publicResource)).start();//生产一个资源线程
        new Thread(new DecreaseRunnable(publicResource)).start();//消费一个资源线程

        new Thread(new IncreaseRunnable(publicResource)).start();//生产一个资源线程
        new Thread(new DecreaseRunnable(publicResource)).start();//消费一个资源线程

        new Thread(new DecreaseRunnable(publicResource)).start();//消费一个资源线程
        new Thread(new IncreaseRunnable(publicResource)).start();//生产一个资源线程

    }
}

实行程序,打印日志如下:

Thread-0---生产了1个,总共有1
Thread-1---消费了1个,总共有0
Thread-1---资源已空,消费者等待...
Thread-2---生产了1个,总共有1
Thread-3---消费了1个,总共有0
Thread-3---资源已空,消费者等待...
Thread-4---生产了1个,总共有1
Thread-5---消费了1个,总共有0
Thread-5---资源已空,消费者等待...

await/signal(signalAll)

JDK5.0以后,Java提供了更加健壮的线程处理机制,包括同步,锁定,线程池等,它们可以实现更细粒度的线程控制。await/signal(signalAll)功能和wait/notify(notifyAll)基本相同,甚至可以完全取代。它们和新引入的锁机制Lock挂钩,具有更大的灵活性,同过在Lock对象上调用newCondition方法,将条件变量和锁对象进行绑定,进而控制并发程序访问竞争资源的安全。
贴上代码,这里只贴上PublicResource类,其它类同上面一样,就不重复贴出:

public class PublicResource {
   private int number = 0;
   private int size = 10;

   private Lock mLock;

   private Condition mIncreaseCondition;
   private Condition mDecreaseCondition;

   public PublicResource() {
       mLock = new ReentrantLock();
       mIncreaseCondition = mLock.newCondition();
       mDecreaseCondition = mLock.newCondition();
   }


   /**
    * 生产
    */
   public int increase() {
       mLock.lock();
       while (number >= size) {
           System.out.println(Thread.currentThread().getName() + "资源已满,生产者等待...");
           try {
               //对象的wait方法被调用后,线程进入对象的等待队列中,并释放对象锁,其它线程可以竞争使用此对象锁,
               //sleep方法使得一个线程进入睡眠状态,但是线程所占有的资源并没有释放。
               mIncreaseCondition.await();
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
       ++number;
       System.out.println(Thread.currentThread().getName() + "---生产者---生产了1个,总共有" + number);
       mIncreaseCondition.signalAll();
       mDecreaseCondition.signalAll();
       mLock.unlock();
       return number;
   }

   /**
    * 消费
    */
   public int decrease() {
       mLock.lock();
       while (number <= 0) {
           System.out.println(Thread.currentThread().getName() + "---资源已空,消费者等待...");
           try {
               mDecreaseCondition.await();
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
       --number;
       System.out.println(Thread.currentThread().getName() + "---消费者---消费了1个,总共有" + number);
       mDecreaseCondition.signalAll();
       mIncreaseCondition.signalAll();
       mLock.unlock();
       return number;
   }

   /**
    * 获取资源大小
    *
    * @return 资源大小
    */
   public int getSize() {
       return size;
   }

}

运行结果如下:

Thread-0---生产者---生产了1个,总共有1
Thread-1---消费者---消费了1个,总共有0
Thread-2---生产者---生产了1个,总共有1
Thread-3---消费者---消费了1个,总共有0
Thread-4---生产者---生产了1个,总共有1
Thread-5---消费者---消费了1个,总共有0
Thread-6---资源已空,消费者等待...
Thread-7---生产者---生产了1个,总共有1
Thread-6---消费者---消费了1个,总共有0

BlockingQueue##

BlockingQueue也是JDK5.0新增内容,在它内部实现了同步的队列,实现方式采用的是await/signal, 在生成对象时指定容量大小,用于阻塞操作是put方法和take方法。
put:类似生产者,容量达到最大时自动阻塞。
take:类似消费者,容量为0时自动阻塞。
贴上PublicResource代码,其它不变,同上。

public class PublicResource {
    private int size = 10;

    private BlockingDeque<Object> mBlockingDeque = new LinkedBlockingDeque<>(10);

    /**
     * 生产
     */
    public int increase() {

        while (mBlockingDeque.size() == size) {
            System.out.println(Thread.currentThread().getName() + "资源已满,生产者等待...");
        }
        try {
            mBlockingDeque.put(new Object());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + "---生产者---生产了1个,总共有" + mBlockingDeque.size());
        return mBlockingDeque.size();
    }

    /**
     * 消费
     */
    public int decrease() {
        while (mBlockingDeque.size() <= 0) {
            System.out.println(Thread.currentThread().getName() + "---资源已空,消费者等待...");
        }
        try {
            mBlockingDeque.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + "---消费者---消费了1个,总共有" + mBlockingDeque.size());
        return mBlockingDeque.size();
    }

    /**
     * 获取资源大小
     *
     * @return 资源大小
     */
    public int getSize() {
        return size;
    }
}

运行结果如下:

Thread-0---生产者---生产了1个,总共有1
Thread-1---消费者---消费了1个,总共有0
Thread-2---生产者---生产了1个,总共有1
Thread-3---消费者---消费了1个,总共有0
Thread-4---生产者---生产了1个,总共有1
Thread-5---消费者---消费了1个,总共有0
Thread-7---生产者---生产了1个,总共有1
Thread-6---资源已空,消费者等待...
Thread-6---消费者---消费了1个,总共有0

推荐阅读更多精彩内容