如何使用信号量解决消费者/生产者任务

Question

我有一个SimpleProducerConsumer类来说明消费者/生产者问题（我不确定它是否正确）。

public class SimpleProducerConsumer { 
    private Stack<Object> stack = new Stack<Object>(); 
    private static final int STACK_MAX_SIZE = 10; 

    public static void main(String[] args) { 
     SimpleProducerConsumer pc = new SimpleProducerConsumer(); 
     new Thread(pc.new Producer(), "p1").start(); 
     new Thread(pc.new Producer(), "p2").start(); 
     new Thread(pc.new Consumer(), "c1").start(); 
     new Thread(pc.new Consumer(), "c2").start(); 
     new Thread(pc.new Consumer(), "c3").start(); 
    } 

    public synchronized void push(Object d) { 
     while (stack.size() >= STACK_MAX_SIZE) 
      try { 
       wait(); 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } 
     try { 
      Thread.sleep(1000); 
     } catch (InterruptedException e) { 
      e.printStackTrace(); 
     } 
     stack.push(new Object()); 
     System.out.println("push " + Thread.currentThread().getName() + " " + stack.size()); 
     notify(); 
    } 

    public synchronized Object pop() { 
     while (stack.size() == 0) 
      try { 
       wait(); 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } 
     try { 
      Thread.sleep(50); 
     } catch (InterruptedException e) { 
      e.printStackTrace(); 
     } 
     stack.pop(); 
     System.out.println("pop " + Thread.currentThread().getName() + " " + stack.size()); 
     notify(); 
     return null; 
    } 

    class Consumer implements Runnable { 
     @Override 
     public void run() { 
      while (true) { 
       pop(); 
      } 
     } 
    } 

    class Producer implements Runnable { 
     @Override 
     public void run() { 
      while (true) { 
       push(new Object()); 
      } 
     } 
    } 
} 

我发现信号量的简单实现here。 （我知道有并发包） 如何更改代码以将java对象的显示器交换到我的自定义信号量？ （要使用信号量说明C/P问题）

信号量：

class Semaphore { 
    private int counter; 

    public Semaphore() { 
     this(0); 
    } 

    public Semaphore(int i) { 
     if (i < 0) 
      throw new IllegalArgumentException(i + " < 0"); 
     counter = i; 
    } 

    public synchronized void release() { 
     if (counter == 0) { 
      notify(); 
     } 
     counter++; 
    } 

    public synchronized void acquire() throws InterruptedException { 
     while (counter == 0) { 
      wait(); 
     } 
     counter--; 
    } 
} 

Answer 1

首先，您应该修复您的测试应用程序。 sleep（）调用是什么？我可以在'生产者'，'消费者'测试线程中理解他们，但为什么他们在推送/流行方法中？此外，测试线程会传输一些对象，但不会检查每个被推入的对象是否只弹出一次。

基于信号量的生产者 - 消费者队列相当平凡。仅基于信号量的有界队列需要三个，一个初始化为零以计算队列中的对象，另一个初始化为队列大小以计算空槽，一个初始化为1以充当互斥体并保护队列免受多址访问。只是Google'生产者消费者'。

测试可以用两个队列和一个具有int计数的消息类来完成。使用顺序计数创建大量对象，并将它们推送到一个队列中，然后启动一个线程负载，将两个队列周围的对象进行混洗。当你停止线程时，（好的，你也需要一个terminate bool），两个队列应该包含所有的原始对象 - 没有额外的，没有短缺和没有重复的。

Answer 2

一般思想从生产者解耦消费者与良好定义的性质的并发中间粘的队列。

Java为此提供了接口BlockingQueue。

您的生产者和消费者唯一需要做的事情是排队上的take()和put()。这也为您提供了可扩展性的优势，因为您可以根据需要添加多个生产者/消费者。