Cách truy xuất và xử lý các ngoại lệ trong ExecutorService của Java

Question

Tôi đang cố gắng tìm ra cách xử lý các ngoại lệ trong cài đặt đa luồng. Tôi muốn thực hiện một số nhiệm vụ song song, mỗi tác vụ có thể ném một ngoại lệ mà tôi cần phải phản ứng (về cơ bản, bằng cách đặt nhiệm vụ thất bại trở lại vào hàng đợi thực hiện). Tuy nhiên, dường như chỉ có cách để thực sự có ngoại lệ từ luồng là tạo ra một tương lai và gọi phương thức get() của nó. Tuy nhiên, điều này về cơ bản biến các cuộc gọi thành các cuộc gọi đồng bộ.

Có lẽ một số mã sẽ minh họa cho điểm:

ExecutorService executor = Executors.newFixedThreadPool(nThreads); 
Task task = taskQueue.poll(); // let's assume that task implements Runnable 
try { 
    executor.execute(task); 
} 
catch(Exception ex) { 
    // record the failed task, so that it can be re-added to the queue 
} 

Tuy nhiên, trong trường hợp này tất cả các nhiệm vụ được đưa ra, nhưng các trường hợp ngoại lệ dường như không bị bắt trong khối catch này ở đây.

Một thay thế sẽ được sử dụng một tương lai thay vì một sợi và lấy kết quả của nó:

try { 
    Future<?> future = executor.submit(task); 
    future.get(); 
} 
... 

Trong trường hợp này, các trường hợp ngoại lệ được bắt ổn trong khối catch, nhưng phải trả giá vì phải chờ đợi cho đến khi thao tác này kết thúc. Vì vậy, các nhiệm vụ được thực thi tuần tự và không song song, như mong muốn.

Tôi đang thiếu gì? Làm thế nào có thể nắm bắt được nhiệm vụ của mỗi Ngoại lệ và phản ứng với họ?

Answer 1

bạn có thể kích hoạt tất cả các nhiệm vụ của bạn trong vòng một vòng lặp và kiểm tra/chờ/thử lại trong một:

Map<Future<?>, Task> futures = new HashMap<Future<?>, Task>() 
while(!taskQueue.isEmpty()){ 
    Task task = taskQueue.poll(); 
    Future<?> future = executor.submit(task); 
    futures.put(future, task); 
} 

for(Map.Entry<Future<?>, Task> entry : futures.entrySet()){ 

    try { 
     entry.getKey().get(); 
    } 
    catch(ExecutionException ex) { 
     // record the failed task, so that it can be re-added to the queue 
     // you should add a retry counter because you want to prevent endless loops 
     taskQueue.add(entry.getValue()); 
    } 
    catch(InterrupredException ex){ 
     // thread interrupted, exit 
     Thread.interrupt(); 
     return; 
    } 
} 

HTH, Mark