Detaljna analiza izvornog koda BlockingQueue

BlockingQueue je interfejs u Javi koji predstavlja nitno-bezbedan red — ne samo da mu može istovremeno pristupati više niti, već dodaje i mehanizam čekanja/obaveštavanja, kako bi se nit koja dobavlja element blokirala kada je red prazan, sve dok red ne postane raspoloživ, ili blokirala nit koja umeće element kada je red pun, sve dok ne postane raspoloživ.

U najčešćem problemu „proizvođač-potrošač“ red se obično posmatra kao kontejner podataka između niti: proizvođač „proizvede“ podatke i stavi ih u kontejner, a potrošač iz kontejnera uzima podatke. Tako su nit proizvođača i nit potrošača razdvojeni i svaka se može posvetiti svom poslu.

Blokirajući red (BlockingQueue) se široko koristi u problemu „proizvođač-potrošač“, jer pruža blokirajuće metode umetanja i uklanjanja. Kada je kontejner reda pun, nit proizvođača se blokira dok red ne postane nepun; kada je kontejner reda prazan, nit potrošača se blokira dok red ne postane neprazan.

Osnovne operacije

Metode koje definiše interfejs BlockingQueue prikazane su ispod:

Pošto BlockingQueue nasleđuje interfejs Queue, ima i osnovne operacije Queue interfejsa, kao što je prikazano:

1) Umetanje elemenata

1. boolean add(E e) : dodaje element na kraj reda; ako je red pun, izbacuje izuzetak IllegalStateException.
2. boolean offer(E e): dodaje element na kraj reda; ako je red pun, vraća false.

2) Brisanje elemenata

1. boolean remove(Object o): uklanja element iz reda; pri uspehu vraća true, a pri neuspehu false.
2. E poll(): dobavlja i uklanja glavu reda; ako je red prazan, vraća null.

3) Pretraga elemenata

1. E element(): dobavlja, ali ne uklanja glavu reda; ako je red prazan, izbacuje izuzetak NoSuchElementException;
2. peek(): dobavlja, ali ne uklanja glavu reda; ako je red prazan, vraća null.

Osim metoda nasleđenih iz Queue interfejsa, BlockingQueue sam definiše i druge metode, na primer operacije umetanja:

1. void put(E e): dodaje element na kraj reda; ako je red pun, nit se blokira dok ne bude prostora.
2. offer(E e, long timeout, TimeUnit unit): umeće zadati element u red; ako je red pun, čeka zadato vreme dok red ne postane raspoloživ.

Na primer operacije uklanjanja:

1. take(): dobavlja i uklanja glavu reda; po potrebi čeka dok red ne postane raspoloživ;
2. poll(long timeout, TimeUnit unit): dobavlja i uklanja glavu reda; ako element nije odmah raspoloživ, čeka zadato vreme.

ArrayBlockingQueue

Klasne implementacije interfejsa BlockingQueue uključuju ArrayBlockingQueue, DelayQueue, LinkedBlockingDeque, LinkedBlockingQueue, LinkedTransferQueue, PriorityBlockingQueue, SynchronousQueue i druge. Krećemo od ArrayBlockingQueue-a.

ArrayBlockingQueue je ograničeni blokirajući red zasnovan na nizu:

• Ograničen: veličina ArrayBlockingQueue određuje se pri konstrukciji i nakon toga se ne može promeniti. Ta granica obezbeđuje kontrolu protoka i pomaže u racionalnom korišćenju resursa.
• FIFO: operacije reda slede princip prvog-unutra-prvi-van.
• Kada je kapacitet reda pun, pokušaj umetanja elementa dovodi do blokiranja; pokušaj uzimanja elementa iz praznog reda takođe blokira.

Treba napomenuti da ArrayBlockingQueue ne garantuje apsolutnu pravednost — pravednost se odnosi na strogo poštovanje reda po apsolutnom vremenu čekanja niti, odnosno nit koja najduže čeka prva pristupa ArrayBlockingQueue-u.

Razlog su drugi faktori na nivou sistema, poput raspoređivanja niti, koji mogu uticati na stvarni redosled izvršavanja. Ako je potreban pravedan ArrayBlockingQueue, pri deklaraciji se flag pravednosti može postaviti na true:

private static ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(10, true);

Polja ArrayBlockingQueue-a su:

/** The queued items */
final Object[] items;

/** items index for next take, poll, peek or remove */
int takeIndex;

/** items index for next put, offer, or add */
int putIndex;

/** Number of elements in the queue */
int count;

/*
 * Concurrency control uses the classic two-condition algorithm
 * found in any textbook.
 */

/** Main lock guarding all access */
final ReentrantLock lock;

/** Condition for waiting takes */
private final Condition notEmpty;

/** Condition for waiting puts */
private final Condition notFull;

• items: niz za čuvanje elemenata reda. Veličina reda se definiše pri konstrukciji i ne menja se tokom životnog veka.
• takeIndex: indeks za sledeću operaciju take, poll, peek ili remove. Pokazuje na poziciju trenutno dostupnog elementa za potrošnju.
• putIndex: indeks za sledeću operaciju put, offer ili add. Pokazuje na poziciju na koju će biti umetnut novi element.
• count: trenutni broj elemenata u redu. Kada dostigne veličinu niza, dalje operacije put će se blokirati.
• lock: objekat ReentrantLock koji štiti pristup redu. Sve operacije pristupa i izmene reda moraju se sinhronizovati preko ove brave.
• notEmpty: uslov Condition za čekanje operacija take. Kada je red prazan, nit koja pokušava da uzme element iz reda čeka na ovom uslovu.
• notFull: uslov Condition za čekanje operacija put. Kada je red pun, nit koja pokušava da doda element u red čeka na ovom uslovu.

Konstruktor je:

public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

1) Detaljno o metodi put

Izvorni kod metode put(E e):

public void put(E e) throws InterruptedException {
    // osigurava da prosleđeni element nije null
    checkNotNull(e);
    final ReentrantLock lock = this.lock;

    // traži bravu; ako je nit prekinuta, izbacuje izuzetak
    lock.lockInterruptibly();
    try {
        // u petlji proverava da li je red pun; ako jeste, čeka na uslovu notFull
        while (count == items.length) {
            notFull.await();
        }
        // red nije pun, umeće element
        enqueue(e);
    } finally {
        // posle try bloka oslobađa bravu, osiguravajući da bude oslobođena
        lock.unlock();
    }
}

Logika metode je jednostavna: kada je red pun (count == items.length), nit se premešta u red čekanja notFull; ako su ispunjeni uslovi za umetanje, direktno se poziva enqueue(e) koja umeće element. Izvorni kod metode enqueue:

private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;
    final Object[] items = this.items;
	//umeće podatak
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
	//obaveštava nit potrošača da u redu ima podataka za potrošnju
    notEmpty.signal();
}

Logika metode enqueue takođe je jednostavna: prvo umeće podatak (items[putIndex] = x), a zatim obaveštava blokirane niti potrošača da u redu sada ima podataka za potrošnju (notEmpty.signal()).

2) Detaljno o metodi take

Izvorni kod metode take:

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
		//ako je red prazan, nema podataka, premešta nit potrošača u red čekanja
        while (count == 0)
            notEmpty.await();
		//dobavlja podatak
        return dequeue();
    } finally {
        lock.unlock();
    }
}

1. Ako je trenutni red prazan, nit potrošača koja dobavlja podatke premešta se u red čekanja;
2. Ako red nije prazan, dobavlja podatke, odnosno završava operaciju izlaska iz reda dequeue. Izvorni kod metode dequeue:

private E dequeue() {
    // assert lock.getHoldCount() == 1;
    // assert items[takeIndex] != null;
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
	//dobavlja podatak
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    //obaveštava blokirane niti proizvođača
	notFull.signal();
    return x;
}

Metod dequeue uglavnom radi dve stvari:

1. Dobavlja podatak iz reda ((E) items[takeIndex]);
2. Obaveštava moguće niti proizvođača koje čekaju umetanje elementa da sada ima raspoloživog prostora, pozivom metode signal uslova notFull.

Iz gornje analize vidimo da metode put i take blokirajuću proizvodnju i potrošnju podataka ostvaruju uglavnom preko mehanizma obaveštavanja uslova Condition.

3) Primer upotrebe

Hajde da pogledamo i primer upotrebe ArrayBlockingQueue-a:

public class ArrayBlockingQueueTest {
    private static ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(10, true);

    public static void main(String[] args) {
        new Thread(new Producer()).start();
        new Thread(new Consumer()).start();
    }

    static class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 100; i++) {
                try {
                    blockingQueue.put(i);
                    System.out.println("Proizvođač proizvodi podatak: " + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    static class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 100; i++) {
                try {
                    Integer data = blockingQueue.take();
                    System.out.println("Potrošač troši podatak: " + data);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

Deo rezultata izvršavanja prikazan je na slici:

LinkedBlockingQueue

LinkedBlockingQueue je nitno-bezbedan blokirajući red zasnovan na povezanoj listi:

• Omogućava efikasno umetanje i brisanje na početku i na kraju reda.
• Kada je red prazan, operacija uzimanja se blokira dok ne postane dostupan novi element. Kada je red pun, operacija umetanja se blokira dok ne bude raspoloživog prostora.
• Maksimalni kapacitet se može zadati pri konstrukciji. Ako se ne zada, podrazumevano je Integer.MAX_VALUE, što znači da je veličina reda ograničena samo raspoloživom memorijom.

Polja LinkedBlockingQueue-a su:

/** Current number of elements */
private final AtomicInteger count = new AtomicInteger();

/**
 * Head of linked list.
 * Invariant: head.item == null
 */
transient Node<E> head;

/**
 * Tail of linked list.
 * Invariant: last.next == null
 */
private transient Node<E> last;

/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();

/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();

/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();

/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();

• count: AtomicInteger koji predstavlja trenutni broj elemenata u redu. Atomskim operacijama se osigurava nitna bezbednost.
• head: čvorna glava reda. Pošto je ovo FIFO red, elementi se uvek uklanjaju s početka. Polje item čvora glave uvek je null; taj čvor služi kao virtuelni, za upravljanje redom.
• last: čvorni rep reda. Novi elementi se uvek umeću na kraj.
• takeLock i putLock: ovo su dve ReentrantLock brave u LinkedBlockingQueue-u. takeLock kontroliše operacije uzimanja, a putLock operacije umetanja. Ovakav dizajn omogućava da se umetanje i uzimanje do izvesne mere paralelno izvršavaju, čime se povećava propusnost reda.
• notEmpty i notFull: dve promenljive uslova Condition, povezane redom sa takeLock i putLock. Kada je red prazan, nit koja pokušava da uzme element iz reda čeka na notEmpty. Kada se novi element smesti u red, te niti koje čekaju bivaju probuđene. Slično, kada je red pun, nit koja pokušava da stavi element u red čeka na notFull, dok ne bude prostora.

Definicija čvora Node povezane liste je:

static class Node<E> {
    E item;

    /**
     * One of:
     * - the real successor Node
     * - this Node, meaning the successor is head.next
     * - null, meaning there is no successor (this is the last node)
     */
    Node<E> next;

    Node(E x) { item = x; }
}

1. item: ovo polje čuva element koji čvor sadrži.

2. next: ovo polje pokazuje na sledbenika čvora u redu. Ima tri moguće vrednosti:

• stvarna referenca na sledbenika;
• referenca na sam čvor, što znači da je sledbenik head.next;
• null, što znači da nema sledbenika, odnosno da je ovaj čvor poslednji u redu.

3. Node(E x): konstruktor čvora, prima element x i dodeljuje ga polju item.

1) Detaljno o metodi put

Izvorni kod metode put:

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    // Note: convention in all put/take/etc is to preset local var
    // holding count negative to indicate failure unless set.
    int c = -1;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
        /*
         * Note that count is used in wait guard even though it is
         * not protected by lock. This works because count can
         * only decrease at this point (all other puts are shut
         * out by lock), and we (or some other waiting put) are
         * signalled if it ever changes from capacity. Similarly
         * for all other uses of count in other wait guards.
         */
		//ako je red pun, blokira trenutnu nit i premešta je u red čekanja
        while (count.get() == capacity) {
            notFull.await();
        }
		//operacija ulaska u red, umetanje podataka
        enqueue(node);
        c = count.getAndIncrement();
		//ako red ispunjava uslov za umetanje, obaveštava blokirane niti proizvođača
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
}

Logika metode put u suštini je ista kao kod ArrayBlockingQueue-a.

1. Provera parametra: ako je prosleđeni element null, izbacuje se NullPointerException. LinkedBlockingQueue ne dozvoljava umetanje null elemenata.

2. Inicijalizacija lokalnih promenljivih:

• int c = -1; čuva broj elemenata reda pre operacije, postavljen na -1 kao znak neuspeha, osim ako se kasnije ne postavi.
• Node<E> node = new Node<E>(e); kreira novi čvor sa elementom e koji treba umetnuti.
• final ReentrantLock putLock = this.putLock; i final AtomicInteger count = this.count; dobavljaju bravu i brojač reda.

3. Pribavljanje brave: putLock.lockInterruptibly(); pokušava da pribavi bravu za operacije umetanja; ako je nit prekinuta, izbacuje InterruptedException.

4. Čekanje da red ne bude pun: ako je red pun (count.get() == capacity), trenutna nit se blokira i čeka da uslov notFull bude ispunjen. Čim bude prostora, nit se budi i nastavlja.

5. Operacija ulaska u red: pozivom enqueue(node); novi čvor se umeće na kraj reda.

6. Ažuriranje brojača: preko c = count.getAndIncrement(); dobavlja i uvećava brojač elemenata reda.

7. Provera i moguće buđenje drugih niti proizvođača: ako red nije pun (c + 1 < capacity), pomoću notFull.signal(); budi druge niti proizvođače koje možda čekaju prostor za umetanje.

8. Oslobađanje brave: finally blok osigurava da brava bude oslobođena po završetku operacije.

9. Moguće buđenje niti potrošača: ako je operacija umetanja prevela red iz praznog u neprazno stanje (c == 0), poziva se signalNotEmpty(); koja budi niti potrošače koje možda čekaju neprazan red.

2) Detaljno o metodi take

Izvorni kod metode take:

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    try {
		//ako je trenutni red prazan, blokira trenutnu nit i premešta je u red čekanja, dok uslov ne bude ispunjen
        while (count.get() == 0) {
            notEmpty.await();
        }
		//uklanja element s početka reda i dobavlja podatak
        x = dequeue();
        c = count.getAndDecrement();
        //ako su ispunjeni uslovi za uklanjanje elementa, obaveštava blokirane niti potrošača
		if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        signalNotFull();
    return x;
}

1. Inicijalizacija lokalnih promenljivih:

• E x; čuva dobavljeni element.
• int c = -1; čuva broj elemenata reda pre operacije, postavljen na -1 kao znak neuspeha, osim ako se kasnije ne postavi.
• final AtomicInteger count = this.count; i final ReentrantLock takeLock = this.takeLock; dobavljaju brojač i bravu reda.

2. Pribavljanje brave: takeLock.lockInterruptibly(); pokušava da pribavi bravu za operacije uzimanja; ako je nit prekinuta, izbacuje InterruptedException.

3. Čekanje da red ne bude prazan: ako je red prazan (count.get() == 0), trenutna nit se blokira i čeka da uslov notEmpty bude ispunjen. Čim red postane neprazan, nit se budi i nastavlja.

4. Operacija izlaska iz reda: pozivom x = dequeue(); uklanja se element s početka reda i dodeljuje promenljivoj x.

5. Ažuriranje brojača: preko c = count.getAndDecrement(); dobavlja se i umanjuje brojač elemenata reda.

6. Provera i moguće buđenje drugih niti potrošača: ako red i dalje ima druge elemente (c > 1), pomoću notEmpty.signal(); budi se druge niti potrošača koje možda čekaju neprazan red.

7. Oslobađanje brave: finally blok osigurava da brava bude oslobođena po završetku operacije.

8. Moguće buđenje niti proizvođača: ako je operacija uzimanja prevela red iz punog u nepuno stanje (c == capacity), poziva se signalNotFull(); koja budi niti proizvođače koje možda čekaju prostor za umetanje.

9. Vraćanje dobavljenog elementa: na kraju se vraća dobavljeni element x.

3) Primer upotrebe

public class LinkedBlockingQueueTest {
    private static LinkedBlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<Integer>(10);

    public static void main(String[] args) {
        new Thread(new Producer()).start();
        new Thread(new Consumer()).start();
    }

    static class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 100; i++) {
                try {
                    blockingQueue.put(i);
                    System.out.println("Proizvođač proizvodi podatak: " + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    static class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 100; i++) {
                try {
                    Integer data = blockingQueue.take();
                    System.out.println("Potrošač troši podatak: " + data);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

Deo rezultata izvršavanja prikazan je na slici:

Poređenje ArrayBlockingQueue i LinkedBlockingQueue

Zajedničko: i ArrayBlockingQueue i LinkedBlockingQueue blokirajuće umetanje i brisanje ostvaruju preko mehanizma obaveštavanja uslova Condition.

Razlike:

1. ArrayBlockingQueue je zasnovan na nizu, a LinkedBlockingQueue na povezanoj listi;
2. ArrayBlockingQueue koristi jedan ReentrantLock za kontrolu pristupa redu, dok LinkedBlockingQueue koristi dve brave (putLock i takeLock) — jednu za operacije umetanja, drugu za operacije uzimanja. Razdvajanje brava odlično odgovara scenarijima u kojima su učestalosti proizvodnje i potrošnje približno iste, jer se proizvodnja i potrošnja mogu izvršavati bez međusobnog ometanja i postići dobra efikasnost.

PriorityBlockingQueue

PriorityBlockingQueue je neograničeni blokirajući red sa svojstvom prioritetnog uređenja. Redosled elemenata u redu prati prirodno uređenje ili prilagođeno uređenje preko prosleđenog komparatora. Prirodno uređenje možete definisati implementacijom interfejsa Comparable.

Kada zadatke treba izvršavati prema prioritetu, PriorityBlockingQueue je vrlo koristan. Sledeći kod pokazuje kako se pomoću PriorityBlockingQueue-a mogu upravljati zadacima različitih prioriteta.

class Task implements Comparable<Task> {
    private int priority;
    private String name;

    public Task(int priority, String name) {
        this.priority = priority;
        this.name = name;
    }

    public int compareTo(Task other) {
        return Integer.compare(other.priority, this.priority); // higher values have higher priority
    }

    public String getName() {
        return name;
    }
}

public class PriorityBlockingQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<>();
        queue.put(new Task(1, "Low priority task"));
        queue.put(new Task(50, "High priority task"));
        queue.put(new Task(10, "Medium priority task"));

        while (!queue.isEmpty()) {
            System.out.println(queue.take().getName());
        }
    }
}

Gornji primer kreira prioritetni blokirajući red i dodaje tri zadatka različitih prioriteta. Oni se uzimaju i ispisuju redom od najvišeg ka nižem prioritetu. Rezultat izvršavanja:

High priority task
Medium priority task
Low priority task

SynchronousQueue

SynchronousQueue je vrlo poseban blokirajući red koji ne čuva nijedan element. Svaka operacija umetanja mora sačekati operaciju uklanjanja druge niti, i obrnuto. Stoga je SynchronousQueue interno zapravo prazan, ali dozvoljava jednoj niti da direktno prenosi element drugoj niti, jedan po jedan.

SynchronousQueue dozvoljava niti da element neposredno preda drugoj niti. Tako, ako jedna nit pokuša umetanje elementa, a postoji druga nit koja pokušava uklanjanje, operacije umetanja i uklanjanja će istovremeno uspeti.

Kada želite da jedna nit direktno pošalje tačnu poruku drugoj niti, možete koristiti SynchronousQueue. Sledeći kod pokazuje komunikaciju među nitima pomoću SynchronousQueue-a:

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

        // Producer Thread
        new Thread(() -> {
            try {
                String event = "SYNCHRONOUS_EVENT";
                System.out.println("Putting: " + event);
                queue.put(event);
                System.out.println("Put successfully: " + event);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();

        // Consumer Thread
        new Thread(() -> {
            try {
                String event = queue.take();
                System.out.println("Taken: " + event);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }
}

Gornji primer kreira SynchronousQueue, u jednoj niti umeće element, a u drugoj niti ga uklanja. Rezultat izvršavanja:

Putting: SYNCHRONOUS_EVENT
Put successfully: SYNCHRONOUS_EVENT
Taken: SYNCHRONOUS_EVENT

LinkedTransferQueue

LinkedTransferQueue je neograničeni transferni red zasnovan na povezanoj listi, koji implementira interfejs TransferQueue i pruža snažan mehanizam komunikacije među nitima. Funkcionalnost mu je slična drugim blokirajućim redovima, ali dodatno sadrži „transfer“ semantiku: dozvoljava prenos elementa direktno od proizvođača ka potrošaču, ako potrošač već čeka. Ako nema potrošača koji čeka, element ulazi u red.

Dve su uobičajene metode:

• transfer(E e), prenosi element potrošaču koji čeka; ako nema potrošača koji čeka, element ulazi u red i blokira dok ne bude potrošen.
• tryTransfer(E e), pokušava da odmah prenese element; ako potrošač čeka, prenos uspeva, inače vraća false.

Ako želite težu kontrolu interakcije između proizvođača i potrošača, možete koristiti LinkedTransferQueue.

public class LinkedTransferQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        LinkedTransferQueue<String> queue = new LinkedTransferQueue<>();

        // Consumer Thread
        new Thread(() -> {
            try {
                System.out.println("Potrošač čeka da dobavi element...");
                String element = queue.take();
                System.out.println("Potrošač primio: " + element);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();

        // Let consumer thread start first
        TimeUnit.SECONDS.sleep(1);

        // Producer Thread
        System.out.println("Proizvođač prenosi element");
        queue.transfer("Hello, World!");

        System.out.println("Proizvođač preneo element");
    }
}

Nit potrošača prvo kreće i čeka na prijem elementa. Nit proizvođača poziva metod transfer i direktno prenosi element potrošaču.

Rezultat izvršavanja:

Potrošač čeka da dobavi element...
Proizvođač prenosi element
Proizvođač preneo element
Potrošač primio: Hello, World!

LinkedBlockingDeque

LinkedBlockingDeque je obostrani blokirajući red zasnovan na povezanoj listi. Podržava umetanje i uklanjanje i s početka i s kraja reda. Zato se LinkedBlockingDeque može koristiti i kao FIFO i kao LIFO red.

Uobičajene metode:

• addFirst(E e), addLast(E e): dodaje element na početak/kraj reda.
• takeFirst(), takeLast(): uklanja i vraća element s početka/kraja reda; ako je red prazan, čeka.
• putFirst(E e), putLast(E e): umeće element na početak/kraj reda; ako je red pun, čeka.
• pollFirst(long timeout, TimeUnit unit), pollLast(long timeout, TimeUnit unit): uklanja i vraća element s početka/kraja reda; ako je red prazan, čeka zadato vreme.

Primer upotrebe:

public class LinkedBlockingDequeDemo {
    public static void main(String[] args) throws InterruptedException {
        LinkedBlockingDeque<String> deque = new LinkedBlockingDeque<>(10);

        // Adding elements at the end of the deque
        deque.putLast("Item1");
        deque.putLast("Item2");

        // Adding elements at the beginning of the deque
        deque.putFirst("Item3");

        // Removing elements from the beginning
        System.out.println(deque.takeFirst()); // Output: Item3

        // Removing elements from the end
        System.out.println(deque.takeLast()); // Output: Item2
    }
}

Rezultat izvršavanja:

Item3
Item2

DelayQueue

DelayQueue je neograničeni blokirajući red za čuvanje elemenata koji implementiraju interfejs Delayed; ti elementi se iz reda mogu uzeti tek kada istekne njihovo kašnjenje. To DelayQueue čini idealnim za realizaciju servisa zakazivanja zasnovanih na vremenskom prioritetu.

Sledeći primer pokazuje kako koristiti DelayQueue.

public class DelayQueueDemo {
    public static void main(String[] args) {
        DelayQueue<DelayedElement> queue = new DelayQueue<>();

        // Stavlja u red element sa kašnjenjem od 5 sekundi
        queue.put(new DelayedElement(5000, "Ovo je element sa kašnjenjem od 5 sekundi"));

        try {
            System.out.println("Uzimam jedan element...");
            // take() će blokirati dok kašnjenje ne istekne
            DelayedElement element = queue.take();
            System.out.println(element.getMessage());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static class DelayedElement implements Delayed {
        private final long delayUntil;
        private final String message;

        public DelayedElement(long delayInMillis, String message) {
            this.delayUntil = System.currentTimeMillis() + delayInMillis;
            this.message = message;
        }

        public String getMessage() {
            return message;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(delayUntil - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            return Long.compare(this.delayUntil, ((DelayedElement) o).delayUntil);
        }
    }
}

Gornji primer kreira DelayQueue i stavlja u njega element sa kašnjenjem od 5 sekundi. Zatim poziva metod take() da bi izvukao element iz reda. Pošto je kašnjenje elementa 5 sekundi, take() će blokirati 5 sekundi dok element ne istekne. Rezultat izvršavanja:

Uzimam jedan element...
Ovo je element sa kašnjenjem od 5 sekundi

Kratak pregled

Ovaj članak predstavlja Java blokirajuće redove: ArrayBlockingQueue, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue, LinkedTransferQueue, LinkedBlockingDeque i DelayQueue. Svi su nitno-bezbedni i mogu se koristiti u višenitnom okruženju.

Blokirajući red je vrlo korisna alatka koja se može koristiti za realizaciju obrasca proizvođač-potrošač ili za komunikaciju među nitima u višenitnom okruženju. Takođe se može koristiti za realizaciju bazena niti i drugih struktura podataka, poput prioritetnog reda, reda sa kašnjenjem i sl.

Princip realizacije blokirajućeg reda je mehanizam obaveštavanja uslova Condition: kada je red prazan, nit potrošača se blokira dok u redu ne bude podataka za potrošnju; kada je red pun, nit proizvođača se blokira dok ne bude raspoloživog prostora.

Blokirajući red je važan pojam u Java konkurentnom programiranju sa širokom primenom u višenitnom programiranju. Stoga treba dobro poznavati načine upotrebe i principe realizacije.

Urednik: Chenmo Wang Er, deo sadržaja potiče iz GitHub repozitorijuma CL0610 https://github.com/CL0610/Java-concurrency.

---

Na GitHub-u je konačno stigao drugi PDF „Mali priručnik o konkurentnom programiranju“open-source baze znanja sa preko 17000 zvezdica „Ergov put ka naprednom Javom“! Obuhvata osnovne pojmove i načine upotrebe niti, memorijski model Java-e, synchronized, volatile, CAS, AQS, ReentrantLock, bazene niti, konkurentne kontejnere, ThreadLocal, model proizvođač-potrošač i druge teme koje su obavezne za intervjue i razvoj — ukupno preko 150 000 reči i preko 200 ručno nacrtanih ilustracija, što se može opisati kao pristupačno i sa humorom... Više detalja: Odlično, Ergov put ka naprednom konkurentnom programiranju.pdf