Structured Concurrency was the second delivery that came out of Project Loom. It was first introduced in Java 19 as a preview API. It aimed to simplify multithreaded programming by treating multiple tasks running in different threads as a single unit of work, thereby streamlining makes the code more readable, streamlining error handling, improving reliability, and enhancing observerability.
In this article, we will look at:
- Look at what structured concurrency is.
- Look at an example using structured concurrency.
- Dive into the underlying API with StructuredTaskScope.
What is Structured Concurrency
Structured concurrency allows developers to reason about concurrent code by defining precise points where execution splits into multiple tasks and where those tasks subsequently merge. It brings the same level of organization to concurrency that for loops and if branches bring to structured programming. It achieves this by restricting the lifespan of concurrent operations to specific scopes. Similar to how variables within a if statement in structured programming are bound to that block, concurrent operations within a structured scope are terminated upon exiting that scope. This ensures that threads do not linger, preventing memory leaks and wasted CPU resources.
Imagine a task that can be broken down into smaller subtasks and those can, in turn, be further divided. This creates a nested, tree-like structure of tasks requiring execution. The entire structure’s lifetime is confined to the scope of the code block in which it is created. In Java’s structured concurrency API, which is encapsulated in the StructuredTaskScope concept we’ll dig into in the following section. Upon closing this scope, it’s assured that all tasks and their respective subtasks have either finished or been canceled.
Basic example of StructuredTaskScope
To demonstrate the fundamental use of StructuredTaskScope, consider this simple example. Keep in mind that this is primarily for educational purposes. Real-world applications frequently involve using specific subclasses that override the handleComplete method or potentially define your custom StructuredTaskScope subclass.
try (var scope = new StructuredTaskScope<String>()) {
Supplier<String> currentSubtask = scope.fork(this::getCurrent);
Supplier<String> forecastSubtask = scope.fork(this::getForecast);
Supplier<String> historySubtask = scope.fork(this::getHistory);
scope.join();
return new Result(currentSubtask.get(),
forecastSubtask.get(),
historySubtask.get());
}
If we put this code into a schema, it would be something like this
To start, we instantiate a StructuredTaskScope within a try-with-resources block. This is crucial as StructuredTaskScope implements AutoClosable, ensuring the close method is automatically called when the block finishes.
We then launch three tasks concurrently using the fork method, which returns a SubTask object. For type safety and clarity, it’s recommended casting this to a Supplier, given that SubTask extends it.
Following this, we invoke join, which blocks the current thread until all tasks within the scope are completed or the scope is shut down.
Once the tasks are finished, we can access their results by calling the get() method on each SubTask.
The core pattern of structured concurrency unfolds as follows:
- Establish the task scope, the root for all subtasks.
- Launch any number of subtasks within this scope.
- The scope’s creator thread joins the scope (and implicitly all subtasks).
- The creator thread blocks until all subtasks finish.
- The creator thread manages any errors.
- The scope closes automatically upon completion.
Subtasks
We need to discuss the SubTask class, specifically its role as the return value of the join methods. While the join methods, due to SubTask‘s implementation of Supplier, effectively return a Supplier, the language designers have purposefully steered developers towards casting the result to Supplier. This design choice encourages treating the result as a data provider, rather than relying on the specific SubTask implementation. It’s important to note that SubTask is a sealed interface.
public sealed interface Subtask<T> extends Supplier<T> permits SubtaskImplThe SubTask interface utilizes a State enumeration to track the completion status of its associated task. This enumeration has three states: UNAVAILABLE, indicating the task is either incomplete or was completed after the scope was shut down; SUCCESS, signifying successful completion with a result available; and FAILED, indicating completion with an exception and no result.
If a SubTask‘s state is SUCCESS, the result can be obtained through the get() method. Attempting to call get() on a SubTask with a state of UNAVAILABLE or FAILED will result in an IllegalStateException. Conversely, if the state is FAILED, the exception thrown during execution can be retrieved using the exception() method. Calling exception() on a SubTask with a state of UNAVAILABLE or SUCCESS will also throw an IllegalStateException.
Structured Concurrency Policies
It’s important to reiterate that direct use of StructuredTaskScope is discouraged. Instead, it serves as a foundation for two subclasses, each implementing a common concurrency design pattern: Invoke All and Invoke Any. These patterns are not novel; they were also implemented in Java 8’s ExecutorService, which utilized collections of Callable objects. However, the ExecutorService‘s execution was subject to variations based on the underlying thread pool’s characteristics. Structured concurrency addresses this by providing predictable task behavior and scope. The same patterns can also be found in the Fork/Join Framework and in CompletableFutures.
InvokeAll
The Invoke All concurrency pattern, which is the more frequently used of the two built-in patterns, facilitates the creation of multiple subtasks and then waits until all subtasks have completed successfully or until any subtask has thrown an exception.
This pattern is provided by the StructuredTaskScope.ShutdownOnFailure static inner class. In the event that any subtask throws an exception, it is assumed that no valid result can be returned to the caller, and the entire scope is immediately shut down, hence the name ShutdownOnFailure. The previous example can be modified to employ the ShutdownOnFailure policy in the following manner:
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { // 1
Supplier<CurrentWeather> currentSubtask = scope.fork(this::getCurrent);// 2
Supplier<Forecast> forecastSubtask = scope.fork(this::getForecast);
Supplier<History> historySubtask = scope.fork(this::getHistory);
scope.join(); // 3
scope.throwIfFailed(); // 4
return new Result(currentSubtask.get(), // 5
forecastSubtask.get(),
historySubtask.get());
}Let us walk through the code above:
- We create a StructuredTaskScope of type ShutdownOnFailure
- We fork three of subtasks
- We call scope.join to make the thread that owns the scope wait until all subtasks are completed (or and exception is thrown)
- Once the thread owning the scope continues, meaning all subtasks are completed or cancelled, we check if an exception was throws via scopeIfFailed
- After this point, we can safely call get() on the subtasks as we can be assured now that no exception was thrown.
Notice how we define the scope as StructuredTaskScope.ShutdownOnSuccess. As said, ShutdownOnSuccess is a static final class defined within StructureTaskScope and extending it.
It uses much of the functionality of StructuredTaskScope, but overrides the protected handleComplete(Subtask) method to add specialized behaviour, that differs between ShutdownOnSuccess and ShutdownOnFailure.
This pattern is typically used if we need to retrieve multiple sets of data related to one another. Imagine a customer details screen that contains the customers address, a list of his outstanding orders, a list of his most recent completed orders, his outstanding balance, and a list of outstanding returns.
Instead of retrieving this information sequentially, we can retrieve it concurrently, while still waiting for all information to be retrieved before continuing. Structured concurrency provides a very easy to follow, sequential flow of code that is easy to read, understand, and reason about.
InvokeAny
The Invoke Any pattern is kind of the opposite compared to the Invoke All pattern. It is a race: the tasks that returns a value first wins. The idea is that multiple tasks are started simultaneously. As soon as any of these tasks returns a result, the other tasks that are still executing will be cancelled. If the first result is an exception, all other tasks will be cancelled as well, and the result of the execution will be this exception.
See the code sample below:
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<>()) { // 1
scope.fork(this::getWeatherData); // 2
scope.fork(this::getOpenWeatherData);
scope.join(); // 3
return scope.result(); // 4
}In this scenario, we aim to fetch weather data from two distinct APIs. Our primary objective is to obtain the data as rapidly as possible, regardless of the API source. Consequently, the returned data might originate from different APIs on each method call. The focus here is on minimizing response time, rather than relying on a particular data provider.
A comparable situation arises when retrieving driving directions from point A to point B. You could query multiple map services, such as Google Maps and Apple Maps, and choose the directions provided by the fastest responding service.
Lets us again step through this code:
- We create a StructuredTaskScope of type ShutdownOnSuccess
- We fork the subtasks. Note that we do not need to hold a reference to the results of the operation
- The scope joins and waits for the first result to be returned
- We return the result.
While resembling the preceding example, this one presents three significant differences. Firstly, it employs a different type of StructuredTaskScope. Secondly, we no longer need to retain references to the SubTask return values during the forking process. The API assumes responsibility for storing the single successful result, thereby eliminating the need for individual subtask queries. Lastly, the scope.result() method provides direct access to the result. If no subtasks complete successfully, an ExecutionException is thrown by default, but this behavior can be customized by overriding the overloaded result method
public <X extends Throwable> T result(Function<Throwable, ? extends X> esf) throws XThis overloaded method accepts a function that generates the desired exception. Notably, as soon as a subtask returns a successful result, the StructuredTaskScope automatically manages the cancellation of all other subtasks, relieving the developer of this responsibility.
Rolling your own strategty
Next to these two default policies, you are able to use your own policy targeted to the rules your business requires. Let us see how can create such a custom policy.
Picture this: you’re planning a ski trip and want to ensure there’s sufficient snow at your destination. We can develop a custom task scope, which we’ll call SkiPolicy, to query a Weather API for snow conditions across a list of candidate cities.
We are using the weatherapi.com REST API here. To access the API you will need an API key. There is a free plan, which limits the number of requests you can you do. Still, 1,000,000 free requests per month should be more than sufficient to find you that ideal skiing resort.
We use the GSon library from Google to convert the JSON responses coming from the API into Java objects.
We start off by defining Java records from the JSON responses
public record Condition(String text) { }
public record Current(Condition condition) { }
public record Location(String name, String country) {}
public record CityWeather(Location location, Current current) { }Next, we define the main method of the application. Will will show this first and the later look at the details of the run() and getForecast() methods.
import com.google.gson.Gson;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.List;
import java.util.stream.Collectors;
public class CustomPolicyExample {
private static final String API_KEY = "Add your api key here";
public static void main(String... args) throws Exception {
if (args.length == 0) {
System.out.println("Usage: java CustomPolicyExample <space-seperated list of cities>");
System.exit(0);
}
new CustomPolicyExample().run(List.of(args));
}
private void run(List<String> cities) throws Exception{
// discussed later
}
private CityWeather getForecast(String city) throws Exception {
// discussed later
}
}The main method takes the cities to examine for the correct skiingn conditions as command line arguments. If there is at lest one parameter specified, the run() method is called, passing on the command line parameters.
private void run(List<String> cities) throws Exception{
try(var scope = new SkiPolicy<CityWeather>()) {
for (var city : cities) {
scope.fork(() -> getForecast(city));
}
scope.join();
scope.results().stream()
.filter(e ->
e.current().condition().text().contains("snow"))
.collect(Collectors.toList())
.forEach(System.out::println);
}
}The run() method forks a different threads for each city to retrieve the weather data. This a done by called the getForecast() method. The run() method than joins the scope, effectively waiting until all threads have completed.
Once completed, the results of all the threads, that are collected in the Scope in the results variable, are filtered for the text “snow”. The ones matching the condition are then printed to console.
For completeness, here is the code from the getForecast method
private CityWeather getForecast(String city) throws Exception {
var uri = "http://api.weatherapi.com/v1/current.json?key="
+ API_KEY
+ "&q="
+ city";
var request = HttpRequest.newBuilder()
.uri(new URI(uri))
.GET()
.build();
try(var client = HttpClient.newHttpClient()) {
var response = client.send(request,
HttpResponse.BodyHandlers.ofString());
CityWeather cw = new Gson().fromJson(response.body(),
CityWeather.class);
return cw;
}
}This URI for the REST call is made of the endpoint address and the parameters for the API key and the name of the city are added.
We then performd simple GET request and upon receiving a result we use the GSon library to convert the JSON to a CityWeather Java object.
What is the future for structured concurrency?
You should now have a fairly good idea what structured concurrency is, how it works, and how you can apply it to your applications.
So should you use it in your production code? Based on functionality, I would certainly say “yes”. It is still a preview function, meaning that the API could still change. It could even be withdrawn completely (as we saw with the StringTemplate JEP), but that is very unlikey.
I had expected the Structured Concurrency API to become final in Java 25, but now I am not so sure in anymore. In fact, it is quite the opposite. There is a new Structural Concurrency JEP (Java Enhancement Proposal) that proposed quite some changes to the API. This proposal (https://openjdk.org/jeps/8340343) is still in draft, but I expect it to target Java 25.
It proposed the use a static open method on StructuredTaskScope and Joiner objects for implementing the policies.
String race(Collection<Callable<String>> tasks) throws InterruptedException {
try (var scope = StructuredTaskScope.open(Joiner.
<String>anySuccessfulResultOrThrow()) {
tasks.forEach(scope::fork);
return scope.join();
}
}Notice the use of the open method. And theuse of the anySuccessfulResultOrThrow method, which is the shutdownOnSuccess from above.
You can again custom implementations by creating custom joiners. For this, you need to implement the Joiner interface:
public interface Joiner<T, R> {
default boolean onFork(Subtask<? extends T> subtask)
default boolean onComplete(Subtask<? extends T> subtask)
R result() throws Throwable
}Conclusion
Structured concurrency is a very useful addition to the standard Java library. It takes the burdon on concurrent programming away from the developer. It lets you write linear code that easy to read, to understand, and to debug. As such, it offers solutions to concurrent programming that so far only could be found in Reactive Programming, but without the problems of Reactive (steep learning curve, added complexity, and hard to debug).
