执行者服务返回不正确的响应

Question

我正在根据学生 ID 从对执行程序服务提交方法的调用列表中创建未来列表。来自服务的响应并未针对所有学生 ID 返回。它运行了正确的次数，但在服务调用中使用的 studentId 要么是第一次，要么是最后一次。它忽略了中间的。请检查下面的代码

private List<Student> studentTask() {
        ExecutorService executor = Executors.newFixedThreadPool(5);
        List<Future<List<Student>>> tasks = new ArrayList<>();
        for(int studentNumber = 1; studentNumber <= 10; studentNumber++) {
            Callable<List<Student>> task = new StudentService(studentNumber);
            Future<List<Student>> recordTask = executor.submit(task);
            tasks.add(recordTask);
        }
        try {
            for (Future<List<Student>> future : tasks) {
                List<Student> student = future.get();
                //...
            }
            return //studentList;
        } catch (Exception e) {
        }
    }

private class StudentService implements Callable<List<Student>>{
   private int studentId;
    
   StudentService(int studentId){
     this.studentId = studentId;
   }
    
   public List<Student> call(){
     return getStudentNames(this.studentId);
   }
 }

public class Student{
   private String studentName;
   private int StudentId;

   //Parameterized Constructor
}

private List<Student> getStudentNames(int studentId){
   List<Student> studentList = // db call returning all student with 
                               // respect to studentId.
   return studentList;
}

在下面的代码中，服务被调用了 10 次，但只有学生 ID 1 和 10。无法获得 2 到 9 的结果，这导致结果不准确。如果我在这里遗漏任何内容，需要帮助理解。

Answer 1

如中所述，您的代码错误到无法编译的地步。而且您省略了一些可能很重要的代码。所以我无法准确诊断你有什么问题。所以我会以我自己的风格修改你的代码，让它工作。

与其尝试填写 Future 个对象的列表，不如填写任务列表，Runnable 或 Callable 个对象。然后您可以调用 ExecutorService#invokeAll 提交所有任务。您会得到一个包含 Future 个对象的列表，以跟踪您提交的任务的完成情况。

首先，让我们将 Student class 定义为 record。

record Student( int number , String name ) { }

在我看来，您在 StudentService class 中混合了两种不同的职责。 class 应该只专注于保存学生的数据。 class 应该 而不是 是 Callable。单独定义Callable，将StudentService对象传给它的构造函数

请注意我们 return 一个 Optional。如果调用程序员提供了一个无效的学生 ID，我们 return 一个空的 Optional 而不是空指针。

class StudentService
{
    private Set < Student > students;

    StudentService ( )
    {
        this.students =
                Set.of(
                        new Student( 1 , "Alice" ) ,
                        new Student( 2 , "Bob" ) ,
                        new Student( 3 , "Carol" ) ,
                        new Student( 4 , "Davis" ) ,
                        new Student( 5 , "Ernestine" ) ,
                        new Student( 6 , "Frank" ) ,
                        new Student( 7 , "Gail" ) ,
                        new Student( 8 , "Harold" ) ,
                        new Student( 9 , "Iris" ) ,
                        new Student( 10 , "Jean-Luc" )
                );
    }

    synchronized Optional < Student > fetchStudentById ( final int id )
    {
        return this.students.stream().filter( student -> student.id() == id ).findAny();
    }
}

注意上面代码中 fetchStudentById 被标记为 synchronized。我们知道这个方法会被跨线程调用。当前的实现可能是 thread-safe，通过 non-modifiable List 流式传输。但是在实际工作中，这个look-up不一定是thread-safe。所以我们把它标记为 synchronized for thread-safety.

如果您对上面代码中看到的流不满意，请知道您可以使用常规循环实现相同的效果。使用流可以使代码更简洁，但此处使用流不重要。

定义我们的任务，Callable 按 ID 查找学生，returnOptional < Student >。我们将用于实际查找学生的 StudentService 对象传递给其构造函数。然后我们传递所需学生的 ID。

class StudentFindingTask implements Callable < Optional < Student > >
{
    private final StudentService studentService;
    private final int studentId;

    public StudentFindingTask ( final StudentService studentService , int studentId )
    {
        this.studentService = studentService;
        this.studentId = studentId;
    }

    @Override
    public Optional < Student > call ( ) throws Exception
    {
        return this.studentService.fetchStudentById( this.studentId );
    }
}

现在我们准备试试这个了。

实例化一个 StudentService 对象以供我们所有的任务使用。

StudentService studentService = new StudentService();

建立任务对象列表。将服务和 ID 传递给每个人。

int limit = 10;
List < StudentFindingTask > tasks = new ArrayList <>( limit );
for ( int studentId = 1 ; studentId <= limit ; studentId++ )
{
    tasks.add( new StudentFindingTask( studentService , studentId ) );
}

准备执行程序服务，以及我们希望它填充的 Future 个对象的列表。

ExecutorService executorService = Executors.newFixedThreadPool( 5 );
List < Future < Optional < Student > > > futures;

将所有这些任务提交给执行程序服务。

try { futures = executorService.invokeAll( tasks ); } catch ( InterruptedException e ) { throw new RuntimeException( e ); }

关闭执行器服务，等待结果。以下是取自 Javadoc 并稍作修改的样板代码。

executorService.shutdown(); // Stop accepting task submissions.
try
{
    if ( ! executorService.awaitTermination( 60 , TimeUnit.SECONDS ) )
    {
        executorService.shutdownNow(); // Cancel currently executing tasks
        if ( ! executorService.awaitTermination( 60 , TimeUnit.SECONDS ) )
        { System.err.println( "Pool did not terminate" ); }
    }
}
catch ( InterruptedException ex )
{
    executorService.shutdownNow();
    Thread.currentThread().interrupt();
}

最后，通过检查每个 Future 对象来报告我们的任务结果。在实际工作中，您可能会询问 future 的完成状态，但我会将其留作 reader.

的练习。

for ( Future < Optional < Student > > future : futures )
{
    try { System.out.println( future.get() ); } catch ( InterruptedException e ) { throw new RuntimeException( e ); } catch ( ExecutionException e ) { throw new RuntimeException( e ); }
}

当运行.

Optional[Student[id=1, name=Alice]]
Optional[Student[id=2, name=Bob]]
Optional[Student[id=3, name=Carol]]
Optional[Student[id=4, name=Davis]]
Optional[Student[id=5, name=Ernestine]]
Optional[Student[id=6, name=Frank]]
Optional[Student[id=7, name=Gail]]
Optional[Student[id=8, name=Harold]]
Optional[Student[id=9, name=Iris]]
Optional[Student[id=10, name=Jean-Luc]]

执行者服务返回不正确的响应

Executor service returning incorrect response

java

multithreading

future

callable

executorservice