具有线程消费者和生产者的循环缓冲区：它获得一些执行

Question

我正在开发具有两个线程：消费者和生产者的循环缓冲区。 我正在使用主动等待Thread.yield。 我知道可以用信号量做到这一点，但我想要没有信号量的缓冲区。

两者都有一个共享变量：bufferCircular。在位置阵列的p

虽然缓冲器未满的有用的信息，producer写入数据，并同时有一些有用的信息consumer在位置阵列的c读取数据。来自BufferCircular的变量nElem是尚未读取的值数据的数量。

该程序运行相当不错9/10次运行。然后，有时候，它会在屏幕上显示最后一个元素（循环数为500）之前在无限循环中得到填充，或者不显示任何元素。

我认为可能是一个liveLock，但我找不到这个错误。

共享变量：

public class BufferCircular { 
    volatile int[] array; 
    volatile int p; 
    volatile int c; 
    volatile int nElem; 

    public BufferCircular(int[] array) { 
     this.array = array; 
     this.p = 0; 
     this.c = 0; 
     this.nElem = 0; 
    } 

    public void writeData (int data) { 
     this.array[p] = data; 
     this.p = (p + 1) % array.length; 
     this.nElem++; 
    } 

    public int readData() { 
     int data = array[c]; 
     this.c = (c + 1) % array.length; 
     this.nElem--; 
     return data; 
    } 

} 

生产者线程：

public class Producer extends Thread { 
    BufferCircular buffer; 
    int bufferTam; 
    int contData; 

    public Productor(BufferCircular buff) { 
     this.buffer = buff; 
     this.bufferTam = buffer.array.length; 
     this.contData = 0; 

    } 

    public void produceData() { 
     this.contData++; 
     this.buffer.writeData(contData); 
    } 

    public void run() { 
     for (int i = 0; i < 500; i++) { 
      while (this.buffer.nElem == this.bufferTam) { 
      Thread.yield(); 
     } 
      this.produceData(); 
     } 
    } 
} 

消费者线程：

public class Consumer extends Thread { 
     BufferCircular buffer; 
     int cont; 

     public Consumer(BufferCircular buff) { 
      this.buffer = buff; 
      this.cont = 0; 
     } 

     public void consumeData() { 
      int data = buffer.readData(); 
      cont++; 
      System.out.println("data " + cont + ": " + data); 
     } 

     public void run() { 
      for (int i = 0; i < 500; i++) { 
       while (this.buffer.nElem == 0) { 
       Thread.yield(); 
       } 
       this.consumeData(); 
      } 
     } 
    } 

主要：

public class Main { 

    public static void main(String[] args) { 
     Random ran = new Random(); 
     int tamArray = ran.nextInt(21) + 1; 
     int[] array = new int[tamArray]; 

     BufferCircular buffer = new BufferCircular(array); 

     Producer producer = new Producer (buffer); 
     Consumer consumer = new Consumer (buffer); 

     producer.start(); 
     consumer.start(); 

     try { 
      producer.join(); 
      consumer.join(); 
     } catch (InterruptedException e) { 
      System.err.println("Error with Threads"); 
      e.printStackTrace(); 
    } 

    } 

} 

任何帮助将受到欢迎。

Answer 1

您的问题在于您的BufferCircular方法对竞争条件敏感。以writeData()为例。它执行3个步骤，其中的一些还没有原子：

this.array[p] = data;    // 1 
this.p = (p + 1) % array.length; // 2 not atomic 
this.nElem++;      // 3 not atomic 

假设2个线程在同一时间进入writeData()。在步骤1中，它们都具有相同的p值，并且都重写了array[p]的值。现在，array[p]被重写两次，并且第一个线程必须写入的数据丢失，因为第二个线程写入同一个索引后。然后他们执行第2步 - 结果是不可预测的，因为p可以增加1或2（p = (p + 1) % array.length由3个操作组成，其中线程可以交互）。然后，第3步。++运算符也不是原子的：它在幕后使用2个操作。所以nElem也增加1或2.

所以我们有完全不可预知的结果。这导致您的程序执行不力。

最简单的解决方案是使readData()和writeData()方法序列化。对于这一点，声明它们​​：

public synchronized void writeData (int data) { //... 
public synchronized void readData() { //... 

---

如果只有一个生产者和一个消费者线程，可以对涉及nElem业务发生竞争条件。解决方案是使用AtomicInteger，而不是int：

final AtomicInteger nElem = new AtomicInteger(); 

，并利用其incrementAndGet()和decrementAndGet()方法。