2

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

两者都有一个共享变量：bufferCircular.

当缓冲区没有充满有用信息时，producer在数组位置写入数据p，当有一些有用信息consumer时，在数组位置读取数据c。变量nElemfromBufferCircular是尚未读取的值数据的数量。

该程序运行良好 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();
    }

    }

}

欢迎任何帮助。

4

1 回答 1

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. ++operator 也不是原子的：它在幕后使用 2 个操作。SonElem也增加 1 或 2。

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

最简单的解决方案是使readData()和writeData()方法序列化。为此，声明它们synchronized：

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

如果您只有一个生产者和一个消费者线程，则可能会在涉及nElem. 解决方案是使用AtomicInteger而不是int：

final AtomicInteger nElem = new AtomicInteger();

并使用它的incrementAndGet()和decrementAndGet()方法。

于 2015-05-03T12:25:55.903 回答