为什么 ArrayBlockingQueue 对象的大小不更新？

Question

我正在关注第 23.6 章中的示例 Proceducer/Consumer 关系：ArrayBlockingQueue（Java - 如何编程第 10 章）。

我试过运行示例，但我不明白为什么 "buffer.size()" 一开始没有更新。

我在Consumer中做了一点修改class。我为 Thread.sleep(Generator.nextInt(300)) 设置了 300 毫秒。这是我的结果。

Producer writes 1   Buffer cells occupied: 0
Consumer reads  1   Buffer cells occupied: 0
Producer writes 2   Buffer cells occupied: 1
Consumer reads  2   Buffer cells occupied: 0
Producer writes 3   Buffer cells occupied: 1
Consumer reads  3   Buffer cells occupied: 0
Producer writes 4   Buffer cells occupied: 1
Consumer reads  4   Buffer cells occupied: 0
Producer writes 5   Buffer cells occupied: 1
Consumer reads  5   Buffer cells occupied: 0
Producer writes 6   Buffer cells occupied: 1
Consumer reads  6   Buffer cells occupied: 0
Producer writes 7   Buffer cells occupied: 1
Consumer reads  7   Buffer cells occupied: 0
Producer writes 8   Buffer cells occupied: 1
Consumer reads  8   Buffer cells occupied: 0
Producer writes 9   Buffer cells occupied: 1
Consumer reads  9   Buffer cells occupied: 0
Producer writes10   Buffer cells occupied: 1
Producer done producing
Terminating producer
Consumer reads 10   Buffer cells occupied: 0

Consumer reads values totaling 55
Terminating Consumer

这是代码。

public class BlockingBufferTest {
public static void main(String[] args) throws InterruptedException {
    ExecutorService executorService = Executors.newCachedThreadPool();

    BlockingBuffer sharedLocation = new BlockingBuffer();

    executorService.execute(new Producer(sharedLocation));
    executorService.execute(new Consumer(sharedLocation));

    executorService.shutdown();
    executorService.awaitTermination(1, TimeUnit.MINUTES);

} // end main method
} // end class

public class BlockingBuffer implements Buffer {

private final ArrayBlockingQueue<Integer> buffer; // shared buffer

public BlockingBuffer() {
    buffer = new ArrayBlockingQueue<>(1);
}
@Override
public void blockingPut(int value) throws InterruptedException {
    buffer.put(value); // place value in buffer
    System.out.printf("%s%2d\t%s%d%n", "Producer writes", value, "Buffer cells occupied: ", buffer.size());
}

@Override
public int blockingGet() throws InterruptedException {
    int readValue = buffer.take(); // remove value from buffer
    System.out.printf("%s %2d\t%s%d%n", "Consumer reads", readValue, "Buffer cells occupied: ", buffer.size());
    return readValue;
}
}

public class Consumer implements Runnable {

private static final SecureRandom generator = new SecureRandom();
private final Buffer sharedLocation; // reference to shared object

public Consumer(Buffer sharedLocation) {
    this.sharedLocation = sharedLocation;
} // end constructor

// stores values 1 to 10 in sharedLocation
@Override
public void run() {
    int sum  = 0;
    for(int count=1; count<=10; count++) {
        try {
            Thread.sleep(generator.nextInt(300));// random sleep
            sum +=sharedLocation.blockingGet();

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    } // end for

    System.out.printf("%n%s %d%n%s%n","Consumer reads values totaling", sum, "Terminating Consumer");
} // end run method
}

public class Producer implements Runnable {

private static final SecureRandom generator = new SecureRandom();
private final Buffer sharedLocation; // reference to shared object

public Producer(Buffer sharedLocation) {
    this.sharedLocation = sharedLocation;
} // end constructor

// stores values 1 to 10 in sharedLocation
@Override
public void run() {
    int sum  = 0;
    for(int count=1; count<=10; count++) {
        try {
            Thread.sleep(generator.nextInt(3000));// random sleep
            sharedLocation.blockingPut(count);
            sum += count;

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    } // end for

    System.out.printf("Producer done producing%nTerminating producer%n");
} // end run method
}

Answer 1

这看起来像是线程排序/抢占式多任务处理的产物。

基本上producer/consumer的每个循环都不是原子的。他们可以在一个循环中互相抢占。

例如，操作顺序可能是这样的：

Producer thread: put value              (size=1)
Consumer thread: read value             (size=0)
Producer thread: x1 = buffer.size() --> x1 = 0
Producer thread: print x1 --> you observe 'Producer writes 1   Buffer cells occupied: 0'
Consumer thread: x2 = buffer.size() --> x2 = 0 
Consumer thread: print x --> you observe 'Consumer reads  1   Buffer cells occupied: 0'

P.S。它实际上比这更棘手一点，所有操作都没有线性排序，只有一些组，但对于这个解释没关系

为什么 ArrayBlockingQueue 对象的大小不更新？

Why does not the size of ArrayBlockingQueue object update?

java

java.util.concurrent