Java Concurrency - อินเทอร์เฟซ BlockingQueue

อินเทอร์เฟซ java.util.concurrent.BlockingQueue เป็นอินเทอร์เฟซย่อยของอินเทอร์เฟซคิวและยังสนับสนุนการดำเนินการเพิ่มเติมเช่นรอให้คิวไม่ว่างเปล่าก่อนที่จะดึงองค์ประกอบและรอให้พื้นที่ว่างในคิวก่อนจัดเก็บองค์ประกอบ .

วิธีการบล็อกคิว

ซีเนียร์	วิธีการและคำอธิบาย
1	boolean add(E e) แทรกองค์ประกอบที่ระบุลงในคิวนี้หากสามารถทำได้ทันทีโดยไม่ละเมิดข้อ จำกัด ด้านความจุคืนค่าจริงเมื่อสำเร็จและโยน IllegalStateException หากไม่มีพื้นที่ว่างในขณะนี้
2	boolean contains(Object o) ส่งคืนจริงหากคิวนี้มีองค์ประกอบที่ระบุ
3	int drainTo(Collection<? super E> c) ลบองค์ประกอบที่มีทั้งหมดออกจากคิวนี้และเพิ่มลงในคอลเลกชันที่กำหนด
4	int drainTo(Collection<? super E> c, int maxElements) ลบองค์ประกอบที่มีอยู่ไม่เกินจำนวนที่กำหนดจากคิวนี้และเพิ่มลงในคอลเล็กชันที่กำหนด
5	boolean offer(E e) แทรกองค์ประกอบที่ระบุลงในคิวนี้หากสามารถทำได้ทันทีโดยไม่ละเมิดข้อ จำกัด ด้านความจุโดยส่งคืนค่าจริงเมื่อสำเร็จและเป็นเท็จหากไม่มีช่องว่างในขณะนี้
6	boolean offer(E e, long timeout, TimeUnit unit) แทรกองค์ประกอบที่ระบุลงในคิวนี้รอจนถึงเวลารอที่ระบุหากจำเป็นเพื่อให้มีพื้นที่ว่าง
7	E poll(long timeout, TimeUnit unit) ดึงและลบส่วนหัวของคิวนี้รอจนถึงเวลารอที่ระบุหากจำเป็นเพื่อให้องค์ประกอบพร้อมใช้งาน
8	void put(E e) แทรกองค์ประกอบที่ระบุลงในคิวนี้รอหากจำเป็นเพื่อให้มีพื้นที่ว่าง
9	int remainingCapacity() ส่งคืนจำนวนขององค์ประกอบเพิ่มเติมที่คิวนี้สามารถทำได้ตามสมควร (ในกรณีที่ไม่มีข้อ จำกัด ของหน่วยความจำหรือทรัพยากร) ยอมรับโดยไม่มีการบล็อกหรือ Integer.MAX_VALUE หากไม่มีขีด จำกัด ภายใน
10	boolean remove(Object o) ลบอินสแตนซ์เดียวขององค์ประกอบที่ระบุออกจากคิวนี้หากมีอยู่
11	E take() ดึงและลบส่วนหัวของคิวนี้รอถ้าจำเป็นจนกว่าองค์ประกอบจะพร้อมใช้งาน

ตัวอย่าง

โปรแกรม TestThread ต่อไปนี้แสดงการใช้อินเตอร์เฟส BlockingQueue ในสภาพแวดล้อมแบบเธรด

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

public class TestThread {

   public static void main(final String[] arguments) throws InterruptedException {
      BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(10);

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

      new Thread(producer).start();
      new Thread(consumer).start();

      Thread.sleep(4000);
   }  


   static class Producer implements Runnable {
      private BlockingQueue<Integer> queue;

      public Producer(BlockingQueue queue) {
         this.queue = queue;
      }

      @Override
      public void run() {
         Random random = new Random();

         try {
            int result = random.nextInt(100);
            Thread.sleep(1000);
            queue.put(result);
            System.out.println("Added: " + result);
            
            result = random.nextInt(100);
            Thread.sleep(1000);
            queue.put(result);
            System.out.println("Added: " + result);
            
            result = random.nextInt(100);
            Thread.sleep(1000);
            queue.put(result);
            System.out.println("Added: " + result);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
      }	   
   }

   static class Consumer implements Runnable {
      private BlockingQueue<Integer> queue;

      public Consumer(BlockingQueue queue) {
         this.queue = queue;
      }
      
      @Override
      public void run() {
         
         try {
            System.out.println("Removed: " + queue.take());
            System.out.println("Removed: " + queue.take());
            System.out.println("Removed: " + queue.take());
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
      }
   }
}

สิ่งนี้จะให้ผลลัพธ์ดังต่อไปนี้

เอาต์พุต

Added: 52
Removed: 52
Added: 70
Removed: 70
Added: 27
Removed: 27