Piszę aplikację audio DSP i zdecydowałem się na model producenta-konsumenta. Dużo o tym czytałemvolatile i innych zagadnień związanych z wątkami, ale mam kilka pytań dotyczących pewnych szczegółów mojego przypadku - w szczególności jedną z rzeczy, które muszę udostępniać między wątkami, jest tablica tablic.

Mam klasę reprezentującą producenta. Aby umożliwić zmianę czasu przetwarzania, sklepy producentan bufory, które będą wypełniać rotację za każdym razem, gdy będzie dostępnych więcej danych audio i przekaże bufor do wątku konsumenta.

Zacznę od moich pytań, a następnie postaram się wyjaśnić mój system w wystarczająco szczegółowy sposób - przepraszam za długi post, dziękuję za przyniesienie mi! Byłbym również bardzo wdzięczny za ogólne uwagi na temat mojego wdrożenia i bezpieczeństwa wątków.

Moje bufory są reprezentowane przez avolatile byte[][] szyk. Doskonale zdaję sobie sprawę, żevolatile tylko sprawia, żeodniesienie niestabilny, ale po przeczytaniu SO i różnych postów na blogu wydaje się, że mam dwie opcje:

mógłbym użyćAtomicIntegerArray. Ale:

Czy obniżę wydajność dla takiej aplikacji?

Czy atomowość jest tym, czego potrzebuję? Zamierzam napisać do całej tablicy za jednym razem,następnie Potrzebuję go, aby był widoczny dla innego wątku, nie potrzebuję każdegoindywidualny napisz, aby był atomowy lub widoczny natychmiast.

Jeśli dobrze rozumiem (np.ten wpis na blogu), zadanie własne, które w moim przypadku jest:buffers[currentBuffer] = buffers[currentBuffer] zapewni publikację, którą zobaczysz w moim kodzie poniżej.

Czy to prawda, że ​​spowoduje to, że wszystkie ostatnie zapisy staną się widoczne?

Czy to działa w przypadku tablicy 2D takiej jak ta?

Spróbuję krótko przejrzeć klasę producenta; są to zmienne instancji:

// 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;

Po uruchomieniuproducerThread iconsumerThread, po prostu wykonują metodyproducerLoop iconsumerLoop odpowiednio.

producerLoop blokuje podczas oczekiwania na dane audio, wczytuje do bufora,wykonuje samodzielne przypisanie do bufora, a następnie używaAtomicInteger przykład do sygnalizowania pętli konsumenta.

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 czeka aż doAtomicInteger buffersToProcess jest większa niż zero, a następnie wywołuje obiekt konsumenta, aby zrobił z danymi wszystko, co chce. PotembuffersToProcess jest zmniejszany i czekamy, aż znów stanie się niezerowy.

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();
  }
}

Wielkie dzięki!