解析生产者与消费者问题

生产者与消费者问题可以被描述为：一个有限缓冲区和两类线程，它们是生产者和消费者，生产者把产品放入缓冲区，相反消费者便是从缓冲区中拿走产品。

生产者在缓冲区满时必须等待，直到缓冲区有空间才继续生产。消费者在缓冲区空时必须等待，直到缓冲区中有产品才能继续读取。

下面用三个实例描述这个问题在线程级别上的Java实现。

1.单生产者-单消费者-单个资源

单个生产者，单个消费者以及缓冲区大小为1（单个资源）的情况，类似于停等协议的实现。

生产者（SingleProducer）：

[1]生产一个资源（Resource）；

[2]notify潜在的等待着的消费者（SingleCostumer）；

[3]wait直到资源被消费。

消费者（SingleCostumer）：

[1]消费一个资源（Resource）；

[2]notify潜在的等待着的生产者；

[3]wait直到资源被产生。

资源Resource.java

package com.zj.one;

 

public class Resource {

    private final int resNum;

 

    public Resource(int resNum) {

       this.resNum = resNum;

    }

 

    public String toString() {

       return "Resource " + resNum;

    }

}

 

生产者SingleProducer.java

package com.zj.one;

import java.util.concurrent.TimeUnit;

 

public class SingleProducer implements Runnable {

    private SinglePlatform platform;

    private int count = 0;

 

    public SingleProducer(SinglePlatform platform) {

       this.platform = platform;

    }

 

    public void run() {

       try {

           while (!Thread.interrupted()) {

              synchronized (this) {

                  while (platform.res != null)

                     wait(); // for the resource to be consumed

              }

              System.out.println("Producer add resource " + ++count);

              synchronized (platform.customer) {

                  platform.res = new Resource(count);

                  platform.customer.notify();

              }

              TimeUnit.MILLISECONDS.sleep(100);

           }

       } catch (InterruptedException e) {

           System.out.println("Producer interrupted");

       }

    }

}

消费者SingleCustomer.java

package com.zj.one;

 

public class SingleCustomer implements Runnable {

    private SinglePlatform platform;

 

    public SingleCustomer(SinglePlatform platform) {

       this.platform = platform;

    }

 

    public void run() {

       try {

           while (!Thread.interrupted()) {

              synchronized (this) {

                  while (platform.res == null)

                     wait(); // ... for the productor to produce

              }

              System.out.println("Customer got " + platform.res);

              synchronized (platform.productor) {

                  platform.res = null;

                  platform.productor.notify(); // Ready for another

              }

           }

       } catch (InterruptedException e) {

           System.out.println("Customer interrupted");

       }

    }

}

单个资源缓冲区SinglePlatform.java

package com.zj.one;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

 

public class SinglePlatform {

    Resource res;

    ExecutorService exec = Executors.newCachedThreadPool();

    SingleProducer productor = new SingleProducer(this);

    SingleCustomer customer = new SingleCustomer(this);

   

    public void runPlatform() throws InterruptedException{

       Future<?> f1=exec.submit(productor);

       Future<?> f2=exec.submit(customer);

       TimeUnit.SECONDS.sleep(3);

       f1.cancel(true);

       f2.cancel(true);

       exec.shutdown();

    }

 

    public static void main(String[] args) throws InterruptedException {

       new SinglePlatform().runPlatform();

    }

}

每次启动程序后，在3秒钟后中断所有的线程，结束程序。

2.单生产者-单消费者-多个资源

当缓冲区长度大于1时，类似于滑动窗口协议。资源类Resource.java不变。缓冲区BufferPlatform.java中使用一个队列来实现滑动窗口。

这里设计两种策略来模拟生产速度快于和慢于消费速度两种情形。策略实现自一个策略接口RunnerStrategy。

生产者与消费者分别持有一个标志位来判断是否需要等待。在队列满时，生产者等待；在队列空时，消费者等待。

缓冲区BufferPlatform.java

package com.zj.two;

 

import java.util.AbstractQueue;

import java.util.concurrent.ConcurrentLinkedQueue;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

 

import com.zj.three.Resource;

 

public class BufferPlatform {

    AbstractQueue<Resource> res = new ConcurrentLinkedQueue<Resource>();

    private int queueSize = 3;//窗口大小

    ExecutorService exec = Executors.newCachedThreadPool();

    RunnerStrategy runnerStrategy;

    SingleProducer producer