Cozi concurente

Cozile care nu se blochează

Implementări sigure pentru fire și, cel mai important, non-blocante pe nodurile conectate.

ConcurrentLinkedQueue<E> - folosește un algoritm fără așteptare adaptat să funcționeze cu colectorul de gunoi. Acest algoritm este destul de eficient și foarte rapid, deoarece este construit pe CAS. Metoda size() poate rula mult timp, deci este mai bine să nu o trageți tot timpul.

ConcurrentLinkedDeque<E> - Deque înseamnă coadă dublă. Aceasta înseamnă că datele pot fi adăugate și extrase din ambele părți. În consecință, clasa acceptă ambele moduri de operare: FIFO (First In First Out) și LIFO (Last In First Out).

În practică, ConcurrentLinkedDeque ar trebui să fie folosit dacă LIFO este absolut necesar, deoarece din cauza bidirecționalității nodurilor, această clasă pierde jumătate din performanță în comparație cu ConcurrentLinkedQueue .

import java.util.concurrent.ConcurrentLinkedQueue;

public class  ConcurrentLinkedQueueExample {
   public static void main(String[] args) {
       ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();

       Thread producer = new Thread(new Producer(queue));
       Thread consumer = new Thread(new Consumer(queue));

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

class Producer implements Runnable {

   ConcurrentLinkedQueue<String> queue;
   Producer(ConcurrentLinkedQueue<String> queue){
       this.queue = queue;
   }
   public void run() {
       System.out.println("Class for adding items to the queue");
       try {
           for (int i = 1; i < 5; i++) {
               queue.add("Item #" + i);
               System.out.println("Added: Item #" + i);
               Thread.sleep(300);
           }
       } catch (InterruptedException ex) {
           ex.printStackTrace();
           Thread.currentThread().interrupt();
       }
   }
}

class Consumer implements Runnable {

   ConcurrentLinkedQueue<String> queue;
   Consumer(ConcurrentLinkedQueue<String> queue){
       this.queue = queue;
   }

   public void run() {
       String str;
       System.out.println("Class for getting items from the queue");
       for (int x = 0; x < 5; x++) {
           while ((str = queue.poll()) != null) {
               System.out.println("Pulled out: " + str);
           }
           try {
               Thread.sleep(600);
           } catch (InterruptedException ex) {
               ex.printStackTrace();
               Thread.currentThread().interrupt();
           }
       }
   }
}

Cozi de blocare

Interfața BlockingQueue<E> - dacă există o mulțime de date, ConcurrentLinkedQueue nu este suficient.

Când firele de execuție nu își fac treaba, puteți obține cu ușurință o excepție OutOfMemmoryException . Și pentru ca astfel de cazuri să nu apară, avem un BlockingQueue pentru lucru cu prezența diferitelor metode de umplere și lucru cu coadă și blocări condiționate.

BlockingQueue nu recunoaște elementele nule și aruncă o excepție NullPointerException atunci când încearcă să adauge sau să obțină un astfel de element. Metoda sondajului returnează un element nul dacă niciun element nu a fost plasat în coadă în timpul expirării.

Implementări BlockingQueue<E>

Să aruncăm o privire mai atentă la fiecare dintre implementările noastre BlockingQueue :

ArrayBlockingQueue<E> este o clasă de coadă de blocare construită pe buffer-ul clasic de inel. Aici avem ocazia să gestionăm „onestitatea” încuietorilor. Dacă fair=false (prestabilit), atunci ordinea firelor nu este garantată.

DelayQueue<E extends Delayed> este o clasă care vă permite să extrageți elemente din coadă numai după o anumită întârziere, definită în fiecare element prin metoda getDelay a interfeței Delayed .

LinkedBlockingQueue<E> este o coadă de blocare pe noduri legate, implementată pe algoritmul „cozi de blocare”: prima blocare este pentru adăugare, a doua este pentru tragerea unui element din coadă. Datorită blocărilor, în comparație cu ArrayBlockingQueue , această clasă are performanțe ridicate, dar necesită mai multă memorie. Dimensiunea cozii este setată prin constructor și este egală cu Integer.MAX_VALUE în mod implicit.

PriorityBlockingQueue<E> este un wrapper cu mai multe fire peste PriorityQueue . Comparatorul este responsabil pentru logica prin care elementul va fi adăugat. Cel mai mic element iese primul.

SynchronousQueue<E> - coada funcționează conform principiului FIFO (first-in-first-out). Fiecare operațiune de inserare blochează firul „Producător” până când firul „Consumator” trage elementul din coadă și invers, „Consumatorul” va aștepta până când „Producătorul” introduce elementul.

BlockingDeque<E> este o interfață care descrie metode suplimentare pentru o coadă de blocare bidirecțională. Datele pot fi introduse și extrase din ambele părți ale cozii.

LinkedBlockingDeque<E> este o coadă de blocare bidirecțională pe nodurile legate, implementată ca o simplă listă bidirecțională cu o singură blocare. Dimensiunea cozii este setată prin constructor și este egală cu Integer.MAX_VALUE în mod implicit.

TransferQueue<E> - interfața este interesantă prin faptul că atunci când un element este adăugat în coadă, este posibil să blocați firul Producer care se inserează până când un alt thread Consumer trage elementul din coadă. De asemenea, puteți adăuga o verificare pentru un anumit timeout sau puteți seta o verificare pentru Consumatorii în așteptare . Ca rezultat, obținem un mecanism de transfer de date cu suport pentru mesaje asincrone și sincrone.

LinkedTransferQueue<E> este o implementare a TransferQueue bazată pe algoritmul Dual Queues with Slack. Folosește intens CAS (vezi mai sus) și parcare fir atunci când este inactiv.