Concurrent Queues

Non-Blocking Queues

Thread-safe va eng muhimi, Queue ning bloklamaydigan amalga oshirilishi (linked nodes) bilan.

ConcurrentLinkedQueue<E> — bu yerda garbage collector bilan ishlashga moslashtirilgan wait-free algoritm ishlatiladi. Bu algoritm juda samarali va juda tez, chunki u CAS asosida qurilgan. size() metodi uzoq vaqt ishlashi mumkin, shunday qilib, uni doim tekshirmaslik yaxshi.

ConcurrentLinkedDeque<E> — Deque ning ma'nosi Double ended queue. Bu shuni anglatadiki, ma'lumotlar ikki tomondan qo'shilishi va chiqarilishi mumkin. Shunga mos ravishda, klass ikkala ish rejimini qo'llab-quvvatlaydi: FIFO (First In First Out) va LIFO (Last In First Out).

Amaliyotda ConcurrentLinkedDeque ni ishlatish kerak bo'lgan LIFO talabi bo'lsa, foydalanish tavsiya etiladi, chunki ikkita yo'naltirilgan nodlar tufayli bu klass ConcurrentLinkedQueue ga nisbatan ishlab chiqishda yarmiga yutqazadi.

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("Elementlarni navbatga qo'shish uchun sinf");
       try {
           for (int i = 1; i < 5; i++) {
               queue.add("Element #" + i);
               System.out.println("Qo'shildi: Element #" + 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("Navbatdan elementlarni olish uchun sinf");
       for (int x = 0; x < 5; x++) {
           while ((str = queue.poll()) != null) {
               System.out.println("Chiqarildi: " + str);
           }
           try {
               Thread.sleep(600);
           } catch (InterruptedException ex) {
               ex.printStackTrace();
               Thread.currentThread().interrupt();
           }
       }
   }
}

Blocking Queues

Interface BlockingQueue<E> — katta miqdordagi ma'lumotlar uchun ConcurrentLinkedQueue yetarli emas.

Agar oqimlar vazifani bajara olmasa, OutOfMemmoryException olishingiz mumkin. Va bunday holatlar yuzaga kelmasligi uchun, bizda BlockingQueue mavjud bo'lib, u turli usullar orqali navbatni to'ldirish va shartlar bo'yicha blokirovkalar bilan ishlashni ta'minlaydi.

BlockingQueue nolga teng bo'lgan elementlarni tan olmaydi va shunday elementlarni qo'shishga yoki olishga urinishda NullPointerException chaqiradi. Nol element poll metodi tomonidan qaytariladi, agar timeout davomida navbatda yangi element joylashtirilmagan bo'lsa.

BlockingQueue<E> Amalga oshirilishi

Keling, bizning BlockingQueue ning har bir amalga oshirilishini batafsil ko'rib chiqaylik:

ArrayBlockingQueue<E> — klassik ring buffer asosida qurilgan blokirovka qiluvchi navbat klassi. Bu yerda bizga "adolat" blokirovkalarni boshqarish imkoniyati mavjud. Agar fair=false (standart), oqimlarning ishlash tartibi kafolatlanmaydi.

DelayQueue<E extends Delayed> — klass, elementlarni faqat belgilangan kechikish o'tganidan so'ng chiqarishga imkon beradi, har bir elementda getDelay metodi orqali Delayed interfeysi orqali aniqlanadi.

LinkedBlockingQueue<E> — ikki yo'nalishli navbatda blokirovka qiluvchi, "two lock queue" algoritmi yordamida amalga oshirilgan: birinchi lock — qo'shish uchun, ikkinchisi — navbatdan chiqarish uchun. Locklar yordamida, ArrayBlockingQueue bilan solishtirganda, bu klass yuqori unumdorlikka ega, lekin unga ko'proq xotira kerak. Navbat hajmi konstruktor orqali belgilanadi va standartda Integer.MAX_VALUE ga teng.

PriorityBlockingQueue<E> — PriorityQueue ustidagi ko'p oqimli qoplama. Comparator elementni qaysi mantiqqa ko'ra qo'shilishini belgilaydi. Birinchi bo'lib navbatdan eng kichik element chiqadi.

SynchronousQueue<E> — navbat FIFO (first-in-first-out) prinsipida ishlaydi. Har bir qo'shish operatsiyasi "Producer" oqimini blokirovka qiladi, toki "Consumer" oqimi navbatdan elementni chiqarmaguncha, va aksincha, "Consumer" "Producer" element qo'shmaguncha kutadi.

BlockingDeque<E> — ikki yo'nalishli blokirovkali navbat uchun qo'shimcha metodlarni tavsiflovchi interfeys. Ma'lumotlarni navbatning ikkala tomoniga qo'shish va chiqarish mumkin.

LinkedBlockingDeque<E> — bir yo'nalishli blokirovkali navbat bog'langan nodalarda, oddiy ikki yo'nalishli ro'yxat sifatida bitta lock bilan amalga oshirilgan. Navbat hajmi konstruktor orqali belgilanadi va standartda Integer.MAX_VALUE ga teng.

TransferQueue<E> — interfeys qiziqarli, chunki navbatga element qo'shilayotganda Producer oqimini blokirovka qilish imkoniyati mavjud Consumer oqimi navbatdan elementni chiqarib olmaguncha. Shuningdek, ma'lum bir timeoutni qo'shish yoki kutayotgan Consumer mavjudligini tekshirishni o'rnatish mumkin. Natijada, biz sinxron va asinxron xabarlar bilan ma'lumotlarni uzatish mexanizmini olamiz.

LinkedTransferQueue<E> — Dual Queues with Slack algoritmiga asoslangan TransferQueue ning amalga oshirilishi. Faol tarzda CAS-ni (yuqoriga qarang) va oqimlarning park qilinishi ularni kutish rejimida ushlayotganini ishlatadi.