BlockingQueue ile Basit Bir Consumer Örneği

Bir BlockingQueue'e girilen elemanları alan en basit tüketici (consumer) aşağıdaki gibi yapılabilir:

import java.util.concurrent.BlockingQueue;

public class SimpleConsumer implements Runnable{

	private final BlockingQueue<Integer> queue;
	
	public SimpleConsumer(BlockingQueue<Integer> queue) {
		this.queue=queue;
	}
	
	@Override
	public void run() {
		
		while (true) {
				
			try {
				
				System.out.println("element bekleniyor ");
				
				int element=queue.take();
				
				System.out.println("element alindi : "+element);
				
			} catch (InterruptedException e) {
				e.printStackTrace();
			}			
		}
	}
}

SimpleConsumer sınıfı bir Thread'dir ve kurucu ile Integer değerleri tutan bir BlockingQueue almaktadır. Sonsuz döngü içinde queue.take() yöntemi ile kuyruğa girilen elemanları beklemektedir. take() yöntemi thread'i kuyruğa yeni bir eleman gelene kadar bekletir. Yeni bir eleman geldiğinde sayıyı alır ve ekrana basar. Sonrasında kuyrukta başka elamanları sırayla alır. Eğer kuyrukta eleman kalmaz ise beklemeye devam eder. SimpleConsumer dışarıdan sonlanmadığı sürece devam edecektir. SimpleConsumer sınıfı test etmek için aşağıdaki gibi bir uygulama yapılabilir:

import java.util.Scanner;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ConsumerBlockingQueueTest {

	public static void main(String[] args) {

		BlockingQueue<Integer> queue = new ArrayBlockingQueue(1000);

		SimpleConsumer consumerTest = new SimpleConsumer(queue);
		Thread t = new Thread(consumerTest);
		t.start();

		boolean exit=false;
		
		try (Scanner scanner = new Scanner(System.in)) {
			
			while(!exit) { // sayi disinda herhangi bir sey girilirse uygulamadan cikilir				
				String input = scanner.next();
				try {
					int number=Integer.parseInt(input);
					queue.add(number);
				} catch (NumberFormatException e) {
					exit=true;
				}				
			}		
		}
		System.out.println("exit");
		
		System.exit(0);
	}
}

İlk olarak en fazla 1000 elemanın olabileceği bir BlockingQueue yaratılır. Ardından SimpleConsumer thread'i başlatılır. Kullanıcıdan sürekli sayı almak için scanner sonsuz döngüye alınır. Alınan sayı queue.add(number) ile kuyruğa eklenir. Kullanıcı sayı dışında bir değer girdiğinde döngü sonlanır ve uygulama kapatılır. Uygulama aşağıdaki gibi test edilebilir :

element bekleniyor 
10
element alindi : 10
element bekleniyor 
25
element alindi : 25
element bekleniyor 
35
element alindi : 35
element bekleniyor 
E
exit

Uygulama ilk açıldığında SimpleConsumer element bekleniyor mesajını yazar ve int element=queue.take(); satırında bekler. 10 değerini girdiğinizde değer kuyruğa eklenir ve SimpleConsumer kuyruktan alır ve element alindi : 10 mesajını basar ve yeni değeri beklemeye başlar. Diğer sayılar içinde aynı durum devam eder. E yazıldığında uygulama sonlanır.