CountDownLatch linh hoạt?

Question

Tôi đã gặp phải sự cố hai lần ngay bây giờ, theo đó một chuỗi nhà sản xuất tạo ra các mục công việc N, gửi chúng đến ExecutorService và sau đó cần đợi cho đến khi tất cả các mục N đã được xử lý.

Hãy cẩn thận

• N là không được biết trước. Nếu tôi chỉ đơn giản là tạo ra một CountDownLatch và sau đó có chủ đề sản xuất await() cho đến khi tất cả công việc được hoàn thành.
• Sử dụng CompletionService không phù hợp vì mặc dù chủ đề sản xuất của tôi cần chặn (tức là bằng cách gọi take()) có không có cách nào báo hiệu rằng tất cả công việc đã hoàn tất.

giải pháp ưa chuộng hiện tại của tôi là sử dụng một bộ đếm số nguyên, và để increment này bất cứ khi nào một mục công việc được gửi và sụt lần nó khi một hạng mục công trình được xử lý. Sau khi đăng ký tất cả các nhiệm vụ của N, luồng sản xuất của tôi sẽ cần phải chờ trên khóa, kiểm tra xem có counter == 0 bất cứ khi nào nó được thông báo hay không. Chủ đề của người tiêu dùng sẽ cần phải thông báo cho nhà sản xuất nếu nó đã giảm giá bộ đếm và giá trị mới là 0.

Có cách tiếp cận tốt hơn cho vấn đề này hoặc có cấu trúc phù hợp trong java.util.concurrent Tôi nên sử dụng hơn là "lăn của riêng tôi"?

Xin cảm ơn trước.

Answer 1

java.util.concurrent.Phaser có vẻ như nó sẽ hoạt động tốt cho bạn. Nó được lên kế hoạch phát hành trong Java 7 nhưng phiên bản ổn định nhất có thể được tìm thấy trên trang web của nhóm lợi ích của jsr166.

Phaser là Thanh chắn tuần hoàn được tôn vinh. Bạn có thể đăng ký số lượng của các bên và khi bạn sẵn sàng chờ đợi trước của họ ở giai đoạn cụ thể.

Một ví dụ nhanh về cách nó sẽ làm việc:

final Phaser phaser = new Phaser(); 

public Runnable getRunnable(){ 
    return new Runnable(){ 
     public void run(){ 
      ..do stuff... 
      phaser.arriveAndDeregister(); 
     } 
    }; 
} 
public void doWork(){ 
    phaser.register();//register self 
    for(int i=0 ; i < N; i++){ 
     phaser.register(); // register this task prior to execution 
     executor.submit(getRunnable()); 
    } 
    phaser.arriveAndAwaitAdvance(); 
} 

Answer 2

Bạn có thể sử dụng tất nhiên một CountDownLatch bảo vệ bởi một AtomicReference để nhiệm vụ của bạn được bao bọc như sau:

public class MyTask extends Runnable { 
    private final Runnable r; 
    public MyTask(Runnable r, AtomicReference<CountDownLatch> l) { this.r = r; } 

    public void run() { 
     r.run(); 
     while (l.get() == null) Thread.sleep(1000L); //handle Interrupted 
     l.get().countDown(); 
    } 
} 

Thông báo rằng nhiệm vụ điều hành công việc của họ và sau đó quay cho đến khi đếm xuống là thiết lập (tức là tổng số nhiệm vụ được biết). Ngay sau khi đếm ngược được thiết lập, họ đếm nó xuống và thoát. Những nhận nộp như sau:

AtomicReference<CountDownLatch> l = new AtomicReference<CountDownLatch>(); 
executor.submit(new MyTask(r, l)); 

Sau điểm tạo/trình công việc của bạn, khi bạn biết có bao nhiêu công việc bạn đã tạo:

latch.set(new CountDownLatch(nTasks)); 
latch.get().await(); 

Answer 3

tôi giả sử sản xuất của bạn không không cần biết khi hàng đợi trống nhưng cần biết khi nào nhiệm vụ cuối cùng đã được hoàn thành.

Tôi sẽ thêm phương thức waitforWorkDone(producer) cho người tiêu dùng. Nhà sản xuất có thể thêm nhiệm vụ N của nó và gọi phương thức chờ. Phương thức chờ chặn luồng đến iff hàng đợi công việc không rỗng và không có nhiệm vụ nào đang thực hiện tại thời điểm này.

Chủ đề người tiêu dùng notifyAll() trên waitfor lock iff nhiệm vụ của nó đã được hoàn thành, hàng đợi trống và không có nhiệm vụ nào khác đang được thực thi.

Answer 4

Tôi đã sử dụng một ExecutorCompletionService cho một cái gì đó như thế này:

ExecutorCompletionService executor = ...; 
int count = 0; 
while (...) { 
    executor.submit(new Processor()); 
    count++; 
} 

//Now, pull the futures out of the queue: 
for (int i = 0; i < count; i++) { 
    executor.take().get(); 
} 

này liên quan đến việc giữ một danh sách các nhiệm vụ đã được gửi đi, vì vậy nếu danh sách của bạn là tùy tiện lâu, phương pháp của bạn có thể được tốt hơn.

Nhưng hãy đảm bảo sử dụng AtomicInteger để điều phối, để bạn có thể tăng nó trong một chuỗi và giảm nó trong chuỗi công việc.

Answer 5

Bạn đã xem cái này:

https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html#invokeAll(java.util.Collection)

Dưới đây là một ví dụ tôi đã viết: https://github.com/moulaali/java_fun/blob/master/src/WaitForAllConcurrencyExample.java

Answer 6

gì bạn được mô tả khá giống với sử dụng Semaphore tiêu chuẩn nhưng được sử dụng 'ngược lại'.

• semaphore của bạn bắt đầu với 0 phép
• Mỗi đơn vị công tác phát hành một giấy phép khi nó hoàn thành
• Bạn chặn bằng cách chờ đợi để có được N cho phép

Thêm vào đó bạn có thể linh động chỉ được M < N giấy phép hữu ích nếu bạn muốn kiểm tra trạng thái trung gian. Ví dụ tôi đang thử nghiệm một hàng đợi thông điệp không đồng bộ nên tôi mong đợi hàng đợi đầy đủ cho một số M < N vì vậy tôi có thể có được M và kiểm tra xem hàng đợi có thực sự đầy đủ không, sau đó lấy các N-M còn lại xếp hàng.