Java—使用公共executor服务实例进行并发处理

Question

我有n个从驱动流中检索记录的工作线程（这对这个问题并不重要），然后将它们推送到执行器服务，在那里处理记录并将其持久化到后端数据库。这个相同的执行器服务实例用于所有工作线程。

现在有一个场景，任何给定的worker循环都会停止处理记录和块，直到它提交的所有记录都得到完全处理。这本质上意味着，对于来自特定工作线程的记录，执行器服务中不应该有挂起/正在运行的线程。

实现的一个非常简单的示例如下：

>

工人阶级

public class Worker {

Worker(Listener listener){
    this.listener = listener;
}

//called periodically to fetch records from a kinesis stream
public void processRecords(Record records) {

    for (Record record : records) {
        listener.handleRecord(record);
    }

    //if 15 minutes has elapsed, run below code. This is blocking.
    listener.blockTillAllRecordsAreProcessed()
}

}

听者类

public class Listener {

    ExecutorService es;

    // same executor service is shared across all listeners.
    Listener(ExecutorService es){
        this.es = es;
    }

    public void handleRecord(Record record) {
        //submit record to es and return
        // non blocking
    }

    public boolean blockTillAllRecordsAreProcessed(){
        // this should block until all records are processed
        // no clue how to implement this with a common es
    }

}

我能想到的唯一方法是为每个工作人员提供本地执行器服务，并为每个批次执行类似调用All的操作，这将稍微改变实现但完成工作。但我觉得应该有更好的方法来解决这个问题。

Answer 1

您可以使用CountdownLatch类来阻止，如下所示：

public void processRecords(List<Record> records) {
 CountDownLatch latch = new CountDownLatch(records.size());
 for (Record record : records) {
     listener.handleRecord(record, latch);
 }

 //if 15 minutes has elapsed, run below code. This is blocking.
 listener.blockTillAllRecordsAreProcessed(latch)
 } 

public class Listener {
 ExecutorService es;
 ...
 public void handleRecord(Record record, CountDownLatch latch) {
     //submit record to es and return
     // non blocking
     es.submit(()->{
        someSyncTask(record);
        latch.countDown();
        
    })
 }

 public boolean blockTillAllRecordsAreProcessed(CountDownLatch latch){
     System.out.println("waiting for processes to complete....");
     try {
          //current thread will get notified if all chidren's are done 
          // and thread will resume from wait() mode.
          latch.await();
          } catch (InterruptedException e) {
               e.printStackTrace();
          }
       }

   }

在这里阅读更多：https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/CountDownLatch.html