java - 2D volatile 数组 : will self-assignment help or do I need AtomicIntegerArray?

Question

我正在编写一个音频 DSP 应用程序，并且选择使用生产者-消费者模型。我读了很多关于volatile的文章和其他线程问题，但我对我的案例的一些细节有几个问题 - 特别是，我需要在线程之间共享的东西之一是数组的数组。

我有一个代表生产者的类。为了允许处理时间的变化，生产者存储 n缓冲区，每当有更多音频数据可用时，它将轮流填充该缓冲区，并将缓冲区传递给消费者线程。

我将从我的问题开始，然后我将尝试足够详细地解释我的系统 - 很抱歉这篇文章很长，感谢您的耐心等待!我也非常感谢有关我的实现及其线程安全性的一般评论。

我的缓冲区由 volatile byte[][] 表示大批。我很清楚volatile只会使引用不稳定，但是阅读了SO和各种博客文章后，似乎我有两个选择:

我可以使用AtomicIntegerArray 。但是:

• 我会牺牲这样的应用程序的性能吗？

• 原子性正是我所需要的吗？我打算一次性写入整个数组，然后我需要它对另一个线程可见，我不需要每个单独写入都是原子的或可见的立即。

如果我理解正确(例如 this blog post )，这是一个 self 分配，在我的例子中是: buffers[currentBuffer] = buffers[currentBuffer]将确保发布，您将在下面的代码中看到这一点。

• 这是否正确，它将导致所有最近的写入变得可见？

• 这对于这样的二维数组有效吗？

<小时/>

我将尝试简要概述生产者类；这些是实例变量:

// The consumer - just an interface with a process(byte[]) method
AudioInputConsumer consumer;

// The audio data source
AudioSource source;

// The number of buffers
int bufferCount;

// Controls the main producer loop
volatile boolean isRunning = false;

// The actual buffers
volatile byte[][] buffers;

// The number of buffers left to process.
// Shared counter - the producer inrements and checks it has not run
// out of buffers, while the consumer decremenets when it processes a buffer
AtomicInteger buffersToProcess = new AtomicInteger(0);

// The producer thread.
Thread producerThread;

// The consumer thread.
Thread consumerThread;

一旦我启动producerThread和consumerThread ，他们只是执行方法 producerLoop和consumerLoop分别。

producerLoop在等待音频数据时阻塞，读入缓冲区，在缓冲区上执行自分配，然后使用AtomicInteger向消费者循环发出信号的实例。

private void producerLoop() {
  int bufferSize = source.getBufferSize();
  int currentBuffer = 0;

  while (isRunning) {
    if (buffersToProcess.get() == bufferCount) {
      //This thread must be faster than the processing thread, we have run out
      // of buffers: decide what to do
      System.err.println("WARNING: run out of buffers");
    }

    source.read(buffers[currentBuffer], 0, bufferSize); // Read data into the buffer
    buffers[currentBuffer] = buffers[currentBuffer];    // Self-assignment to force publication (?)
    buffersToProcess.incrementAndGet();                 // Signal to the other thread that there is data to read
    currentBuffer = (currentBuffer + 1) % bufferCount;  // Next buffer
  }
}

consumerLoop等到AtomicInteger buffersToProcess大于零，然后调用消费者对象对数据执行任何它想要的操作。之后buffersToProcess被递减，我们等待它再次变为非零。

private void consumerLoop() {
  int currentBuffer = 0;

  while (isRunning) {
    if (buffersToProcess.get() > 0) {
      consumer.process(buffers[currentBuffer]);          // Process the data
      buffersToProcess.decrementAndGet();                // Signal that we are done with this buffer
      currentBuffer = (currentBuffer + 1) % bufferCount; // Next buffer
    }
    Thread.yield();
  }
}

非常感谢!

Answer 1

您确实需要原子性，因为写入数组是一个非原子过程。具体来说，Java 肯定永远不会保证对数组成员的写入对其他线程不可见，直到您选择发布它们为止。

一种选择是每次创建一个新数组，对其进行完全初始化，然后通过 volatile 发布，但这可能会产生巨大的成本，因为 Java 坚持新分配的数组必须是首先归零，并且由于 GC 开销。您可以通过“双缓冲”方案来克服这个问题，在该方案中您只保留两个数组并在它们之间进行切换。这种方法有其危险:线程可能仍在从您的写入线程已标记为非 Activity 线程的数组中读取。这在很大程度上取决于代码的精确细节。

唯一的其他选择是在经典、无聊的同步 block 中完成整个读写。这样做的优点是延迟非常可预测。就我个人而言，如果确实受到实际性能问题的压力，我会从这里开始，然后继续进行任何更复杂的事情。

您还可以使用读写锁进行锁定，但这只有在多个线程同时读取数组时才会有效。这似乎不是你的情况。