Làm thế nào để ngắt một BlockingQueue đang chặn trên take()?

Question

Tôi có một lớp học lấy đối tượng từ một số BlockingQueue và xử lý chúng bằng cách gọi số take() trong vòng lặp liên tục. Tại một số điểm tôi biết rằng không có nhiều đối tượng sẽ được thêm vào hàng đợi. Tôi làm cách nào để gián đoạn phương thức take() để nó dừng chặn?

Dưới đây là lớp mà xử lý các đối tượng:

public class MyObjHandler implements Runnable { 

    private final BlockingQueue<MyObj> queue; 

    public class MyObjHandler(BlockingQueue queue) { 
    this.queue = queue; 
    } 

    public void run() { 
    try { 
     while (true) { 
     MyObj obj = queue.take(); 
     // process obj here 
     // ... 
     } 
    } catch (InterruptedException e) { 
     Thread.currentThread().interrupt(); 
    } 
    } 
} 

Và đây là phương pháp sử dụng lớp này để xử lý các đối tượng:

public void testHandler() { 

    BlockingQueue<MyObj> queue = new ArrayBlockingQueue<MyObj>(100); 

    MyObjectHandler handler = new MyObjectHandler(queue); 
    new Thread(handler).start(); 

    // get objects for handler to process 
    for (Iterator<MyObj> i = getMyObjIterator(); i.hasNext();) { 
    queue.put(i.next()); 
    } 

    // what code should go here to tell the handler 
    // to stop waiting for more objects? 
} 

Answer 1

Nếu làm gián đoạn các chủ đề không phải là một lựa chọn, một là đặt một "dấu hiệu" hoặc "lệnh" đối tượng trên hàng đợi đó sẽ được công nhận như vậy bởi MyObjHandler và nghỉ ngơi ra khỏi vòng lặp.

Answer 2

BlockingQueue<MyObj> queue = new ArrayBlockingQueue<MyObj>(100); 
MyObjectHandler handler = new MyObjectHandler(queue); 
Thread thread = new Thread(handler); 
thread.start(); 
for (Iterator<MyObj> i = getMyObjIterator(); i.hasNext();) { 
    queue.put(i.next()); 
} 
thread.interrupt(); 

Tuy nhiên, nếu bạn làm điều này, thread có thể bị gián đoạn trong khi vẫn còn các mục trong hàng đợi, đang chờ xử lý. Bạn có thể muốn xem xét sử dụng poll thay vì take, điều này sẽ cho phép chuỗi xử lý hết thời gian chờ và chấm dứt khi quá trình chờ trong một thời gian không có đầu vào mới.

Answer 3

Ngắt thread:

thread.interrupt() 

Answer 4

Hoặc đừng làm gián đoạn, khó chịu của nó.

public class MyQueue<T> extends ArrayBlockingQueue<T> { 

     private static final long serialVersionUID = 1L; 
     private boolean done = false; 

     public ParserQueue(int capacity) { super(capacity); } 

     public void done() { done = true; } 

     public boolean isDone() { return done; } 

     /** 
     * May return null if producer ends the production after consumer 
     * has entered the element-await state. 
     */ 
     public T take() throws InterruptedException { 
      T el; 
      while ((el = super.poll()) == null && !done) { 
       synchronized (this) { 
        wait(); 
       } 
      } 

      return el; 
     } 
    } 

1. khi puts sản xuất đối tượng vào hàng đợi, hãy gọi queue.notify(), nếu nó kết thúc, hãy gọi queue.done()
2. vòng lặp while (! Queue.isDone() ||! Queue.isEmpty())
3. nghiệm thu() giá trị trả for null

Answer 5

Rất muộn nhưng Hope this helps khác nữa như tôi phải đối mặt với vấn đề tương tự và sử dụngCách tiếp cận 0 được đề xuất bởi erickson above với một số thay đổi nhỏ,

class MyObjHandler implements Runnable 
{ 
    private final BlockingQueue<MyObj> queue; 
    public volatile boolean Finished; //VOLATILE GUARANTEES UPDATED VALUE VISIBLE TO ALL 
    public MyObjHandler(BlockingQueue queue) 
    { 
     this.queue = queue; 
     Finished = false; 
    } 
    @Override 
    public void run() 
    {   
     while (true) 
     { 
      try 
      { 
       MyObj obj = queue.poll(100, TimeUnit.MILLISECONDS); 
       if(obj!= null)//Checking if job is to be processed then processing it first and then checking for return 
       { 
        // process obj here 
        // ... 
       } 
       if(Finished && queue.isEmpty()) 
        return; 

      } 
      catch (InterruptedException e) 
      {     
       return; 
      } 
     } 
    } 
} 

public void testHandler() 
{ 
    BlockingQueue<MyObj> queue = new ArrayBlockingQueue<MyObj>(100); 

    MyObjHandler handler = new MyObjHandler(queue); 
    new Thread(handler).start(); 

    // get objects for handler to process 
    for (Iterator<MyObj> i = getMyObjIterator(); i.hasNext();) 
    { 
     queue.put(i.next()); 
    } 

    // what code should go here to tell the handler to stop waiting for more objects? 
    handler.Finished = true; //THIS TELLS HIM 
    //If you need you can wait for the termination otherwise remove join 
    myThread.join(); 
} 

này giải quyết tất cả những vấn đề

1. Dán cờ hiệu các BlockingQueue để nó biết nó đã không phải chờ đợi nhiều hơn cho các yếu tố
2. Không bị gián đoạn ở giữa để các khối xử lý chấm dứt chỉ khi tất cả các mục trong hàng đợi được xử lý và không còn mục nào được thêm vào