Structured Concurrency in Java is an improvement of Java from Java 19, on implementing and maintaining code related to the execution of tasks consisting of many sub-tasks handled by multi-threading. This improvement makes our code easier to read, and controls how the execution of subtasks will take place, and if an error occurs during the execution of subtasks, how we will handle it. How is it in detail? We will find out together in this tutorial!
As an example, I will create a new Maven project as follows:
Suppose my example application has the task of retrieving student information from many different sources, maybe from a database or from a web service. Because these sources are independent, we can use Java’s multi-threading to submit corresponding subtasks to each source!
I have a Student class containing simple student information as follows:
|
1 2 3 4 5 |
package com.huongdanjava.java; public record Student(String name) { } |
Class StudentDatabaseService implements the Callable interface to retrieve student information from the database with simple content as follows:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
package com.huongdanjava.java; import java.util.Arrays; import java.util.List; import java.util.concurrent.Callable; public class StudentDatabaseService implements Callable<List<Student>> { @Override public List<Student> call() throws Exception { return Arrays.asList(new Student("Khanh"), new Student("Quan")); } } |
The StudentWebService class also implements the Callable interface, and has simple content as follows:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
package com.huongdanjava.java; import java.util.Arrays; import java.util.List; import java.util.concurrent.Callable; public class StudentWebService implements Callable<List<Student>> { @Override public List<Student> call() throws Exception { return Arrays.asList(new Student("Thanh"), new Student("Binh")); } } |
To execute these subtasks using concurrency, I will create a new thread pool using the Executor Service framework and submit these subtasks to get the following results:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
package com.huongdanjava.java; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; public class Application { public static void main(String[] args) throws Exception { try (ExecutorService executorService = Executors.newFixedThreadPool(10)) { Future<List<Student>> studentsFromDatabaseFuture = executorService.submit( new StudentDatabaseService()); Future<List<Student>> studentsFromWebServiceFuture = executorService.submit( new StudentWebService()); List<Student> studentsFromDatabase = studentsFromDatabaseFuture.get(); List<Student> studentsFromWebService = studentsFromWebServiceFuture.get(); List<Student> students = new ArrayList<>(); students.addAll(studentsFromDatabase); students.addAll(studentsFromWebService); students.forEach(s -> System.out.println(s.name())); } } } |
The result when running the application will be as follows:
This is a happy case, subtasks proceed normally without any errors.
In case a sub-task fails due to some reason, for example:
|
1 2 3 4 5 6 7 8 9 10 11 12 |
package com.huongdanjava.java; import java.util.List; import java.util.concurrent.Callable; public class StudentDatabaseService implements Callable<List<Student>> { @Override public List<Student> call() throws Exception { throw new Exception("Something went wrong"); } } |
the remaining tasks take time to complete:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
package com.huongdanjava.java; import java.util.Arrays; import java.util.List; import java.util.concurrent.Callable; public class StudentWebService implements Callable<List<Student>> { @Override public List<Student> call() throws Exception { Thread.sleep(10000); return Arrays.asList(new Student("Thanh"), new Student("Binh")); } } |
When you run the example again, you will see that even though one sub-task has an error, our program will wait until the remaining task finishes running for 10 seconds before throwing the error:
How can we cancel the remaining subtasks to run all these tasks again instead of waiting for the remaining subtasks to complete with results that are not what we expected?
For example, you only need the results of one of the subtasks. There is no need to run all subtasks, how can we cancel other subtasks when one of the subtasks has been completed?
To solve the above questions, you can use Java’s Structured Concurrency!
You can use Java’s StructuredTaskScope class to implement this Structured Concurrency. This StructuredTaskScope class will help us clearly define the scope of each subtask, easily cancel other subtasks when an exception occurs in a subtask. We can also easily cancel other subtasks if their results are no longer needed.
With the example above, I can rewrite the code using the StructuredTaskScope class as follows:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
package com.huongdanjava.java; import java.util.ArrayList; import java.util.List; import java.util.concurrent.StructuredTaskScope; import java.util.concurrent.StructuredTaskScope.Subtask; public class Application { public static void main(String[] args) throws Exception { try (var scope = new StructuredTaskScope<>()) { Subtask<List<Student>> studentsFromDatabaseSubtask = scope.fork( new StudentDatabaseService()); Subtask<List<Student>> studentsFromWebServiceSubtask = scope.fork( new StudentWebService()); scope.join(); List<Student> studentsFromDatabase = studentsFromDatabaseSubtask.get(); List<Student> studentsFromWebService = studentsFromWebServiceSubtask.get(); List<Student> students = new ArrayList<>(); students.addAll(studentsFromDatabase); students.addAll(studentsFromWebService); students.forEach(s -> System.out.println(s.name())); } } } |
We will replace the ExecutorService class with the StructuredTaskScope class. Each sub-task will be a Subtask, we use the fork() method of the StructuredTaskScope class to submit the task.
In addition, you also need to call the join() method of the StructuredTaskScope class so that the program waits for the tasks to be completed. The results of each task can be retrieved using the get() method just like when we use the Future interface!
In the happy case, with:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
package com.huongdanjava.java; import java.util.Arrays; import java.util.List; import java.util.concurrent.Callable; public class StudentDatabaseService implements Callable<List<Student>> { @Override public List<Student> call() throws Exception { return Arrays.asList(new Student("Khanh"), new Student("Quan")); } } |
and:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
package com.huongdanjava.java; import java.util.Arrays; import java.util.List; import java.util.concurrent.Callable; public class StudentWebService implements Callable<List<Student>> { @Override public List<Student> call() throws Exception { return Arrays.asList(new Student("Thanh"), new Student("Binh")); } } |
you will see the same results as when we use the ExecutorService class with the Future interface:
The StructuredTaskScope class defines default Policies that allow us to define the behavior we want when an error occurs in a subtask, or if a subtask is completed and we don’t need to run other subtasks.
There are 2 default Policies that the StructuredTaskScope class has defined:
- ShutdownOnFailure
- ShutdownOnSuccess
The ShutdownOnSuccess policy will cancel other subtasks if one of the subtasks has been completed. You can initialize object of the StructuredTaskScope class with the ShutdownOnSuccess policy as follows:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
package com.huongdanjava.java; import java.util.ArrayList; import java.util.List; import java.util.concurrent.StructuredTaskScope; public class Application { public static void main(String[] args) throws Exception { try (var scope = new StructuredTaskScope.ShutdownOnSuccess<List<Student>>()) { scope.fork(new StudentDatabaseService()); scope.fork(new StudentWebService()); scope.join(); List<Student> students = new ArrayList<>(); students.addAll(scope.result()); students.forEach(s -> System.out.println(s.name())); } } } |
In this case, we no longer need to call the get() method to get the results for each subtask. We just need to use the result() method of the StructuredTaskScope class!
The result you will now see is as follows:
After completing the StudentWebService subtask, the results have been returned!
The ShutdownOnFailure policy will cancel other subtasks if an exception occurs in a subtask. In this case, you will write code as follows:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 |
package com.huongdanjava.java; import java.util.ArrayList; import java.util.List; import java.util.concurrent.StructuredTaskScope; import java.util.concurrent.StructuredTaskScope.Subtask; public class Application { public static void main(String[] args) throws Exception { try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { Subtask<List<Student>> studentsFromDatabaseSubtask = scope.fork( new StudentDatabaseService()); Subtask<List<Student>> studentsFromWebServiceSubtask = scope.fork( new StudentWebService()); scope.join(); scope.throwIfFailed(); List<Student> studentsFromDatabase = studentsFromDatabaseSubtask.get(); List<Student> studentsFromWebService = studentsFromWebServiceSubtask.get(); List<Student> students = new ArrayList<>(); students.addAll(studentsFromDatabase); students.addAll(studentsFromWebService); students.forEach(s -> System.out.println(s.name())); } } } |
We will initialize the object of the StructuredTaskScope class with the ShutdownOnFailure policy. The throwIfFailed() method of the ShutdownOnFailure class will take the exception that occurs in the sub-task to throw!
Rerun the example with:
|
1 2 3 4 5 6 7 8 9 10 11 12 |
package com.huongdanjava.java; import java.util.List; import java.util.concurrent.Callable; public class StudentDatabaseService implements Callable<List<Student>> { @Override public List<Student> call() throws Exception { throw new Exception("Something went wrong"); } } |
and:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
package com.huongdanjava.java; import java.util.Arrays; import java.util.List; import java.util.concurrent.Callable; public class StudentWebService implements Callable<List<Student>> { @Override public List<Student> call() throws Exception { Thread.sleep(10000); return Arrays.asList(new Student("Thanh"), new Student("Binh")); } } |
you will see our program throw an error right after running, instead of having to wait 10 seconds for the StudentWebService task to complete.
So in this article, I have introduced you to Structured Concurrency in Java, and the problems that Structured Concurrency has solved.
