You can't ruin Java with a thread: Part IV - Callable, Future and friends

Introduction

We have already looked at how threads are created in the first part . Let's remember again. A thread is Threadsomething that runs in it run, so let’s use the tutorialspoint java online compiler and execute the following code:

public class HelloWorld {

    public static void main(String []args){
        Runnable task = () -> {
            System.out.println("Hello World");
        };
        new Thread(task).start();
    }
}

Is this the only option for running a task in a thread?

java.util.concurrent.Callable

It turns out that java.lang.Runnable has a brother and his name is java.util.concurrent.Callable and he was born in Java 1.5. What are the differences? If we take a closer look at the JavaDoc of this interface, we see that, unlike Runnable, the new interface declares a method callthat returns a result. Also, by default it throws Exception. That is, it saves us from the need to write try-catchblocks for checked exceptions. Not bad already, right? Now we have Runnablea new task instead:

Callable task = () -> {
	return "Hello, World!";
};

But what to do with it? Why do we even need a task running on a thread that returns a result? Obviously, in the future we expect to receive the result of actions that will be performed in the future. Future in English - Future. And there is an interface with exactly the same name:java.util.concurrent.Future

java.util.concurrent.Future

The java.util.concurrent.Future interface describes an API for working with tasks whose results we plan to obtain in the future: methods for obtaining results, methods for checking status. We Futureare interested in its implementation java.util.concurrent.FutureTask . That is Task, this is what will be executed in Future. What is also interesting about this implementation is that it implements and Runnable. You can consider this a kind of adapter of the old model of working with tasks in threads and the new model (new in the sense that it appeared in java 1.5). Here's an example:

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

public class HelloWorld {

    public static void main(String []args) throws Exception {
        Callable task = () -> {
            return "Hello, World!";
        };
        FutureTask<String> future = new FutureTask<>(task);
        new Thread(future).start();
        System.out.println(future.get());
    }
}

As can be seen from the example, using the method we obtain getthe result from the problem task. (!)Important, that at the moment the result is obtained using the method, getexecution becomes synchronous. What mechanism do you think will be used here? That's right, there is no synchronization block - therefore we will see WAITING in JVisualVM not as monitoror wait, but as the very same one park(since the mechanism is used LockSupport).

Functional Interfaces

Next we will talk about classes from Java 1.8, so it would be useful to make a brief introduction. Let's look at the following code:

Supplier<String> supplier = new Supplier<String>() {
	@Override
	public String get() {
		return "String";
	}
};
Consumer<String> consumer = new Consumer<String>() {
	@Override
	public void accept(String s) {
		System.out.println(s);
	}
};
Function<String, Integer> converter = new Function<String, Integer>() {
	@Override
	public Integer apply(String s) {
		return Integer.valueOf(s);
	}
};

There's a lot of unnecessary code, isn't there? Each of the declared classes performs a single function, but to describe it we use a bunch of unnecessary auxiliary code. And the Java developers thought so too. Therefore, they introduced a set of “functional interfaces” ( @FunctionalInterface) and decided that now Java itself will “think out” everything for us, except the important ones:

Supplier<String> supplier = () -> "String";
Consumer<String> consumer = s -> System.out.println(s);
Function<String, Integer> converter = s -> Integer.valueOf(s);

Supplier- provider. It has no parameters, but it returns something, that is, it supplies it. Consumer- consumer. It takes something as input (parameter s) and does something with it, that is, consumes something. There is another function. It takes something as input (parameter s), does something and returns something. As we see, generics are actively used. If you are unsure, you can remember them and read “ The theory of generics in Java or how to put parentheses in practice .”

CompletableFuture

As time went on, Java 1.8 introduced a new class called CompletableFuture. It implements the interface Future, meaning ours taskwill be executed in the future and we can execute getand get the result. But he also implements some CompletionStage. From the translation its purpose is already clear: it is a certain Stage of some kind of calculation. A brief introduction to the topic can be found in the overview " Introduction to CompletionStage and CompletableFuture ". Let's get straight to the point. Let's look at the list of available static methods to help us get started: Here are the options for using them:

import java.util.concurrent.CompletableFuture;
public class App {
    public static void main(String []args) throws Exception {
        // CompletableFuture уже содержащий результат
        CompletableFuture<String> completed;
        completed = CompletableFuture.completedFuture("Просто meaning");
        // CompletableFuture, запускающий (run) новый поток с Runnable, поэтому он Void
        CompletableFuture<Void> voidCompletableFuture;
        voidCompletableFuture = CompletableFuture.runAsync(() -> {
            System.out.println("run " + Thread.currentThread().getName());
        });
        // CompletableFuture, запускающий новый поток, результат которого возьмём у Supplier
        CompletableFuture<String> supplier;
        supplier = CompletableFuture.supplyAsync(() -> {
            System.out.println("supply " + Thread.currentThread().getName());
            return "Значение";
        });
    }
}

If we run this code, we will see that creation CompletableFutureinvolves starting the entire chain. Therefore, while there is some similarity with the SteamAPI from Java8, this is the difference between these approaches. For example:

List<String> array = Arrays.asList("one", "two");
Stream<String> stringStream = array.stream().map(value -> {
	System.out.println("Executed");
	return value.toUpperCase();
});

This is an example of the Java 8 Stream Api (you can read more about it here " Java 8 Stream API Guide in Pictures and Examples "). If you run this code, it Executedwill not display. That is, when creating a stream in Java, the stream does not start immediately, but waits until a value is needed from it. But CompletableFutureit starts the chain for execution immediately, without waiting for it to be asked for the calculated value. I think it's important to understand this. So we have CompletableFuture. How can we create a chain and what means do we have? Let's remember about the functional interfaces that we wrote about earlier.

• We have a function ( Function) that takes A and returns B. It has a single method - apply(apply).
• We have a consumer ( Consumer) that accepts A and returns nothing ( Void ). It has only one method - accept(accept).
• We have code running on a thread Runnablethat does not accept or return. It has only one method - run(run).

The second thing to remember is that CompletalbeFuturein its work it uses Runnableconsumers and functions. Given this, you can always remember that you CompletableFuturecan do this:

public static void main(String []args) throws Exception {
        AtomicLong longValue = new AtomicLong(0);
        Runnable task = () -> longValue.set(new Date().getTime());
        Function<Long, Date> dateConverter = (longvalue) -> new Date(longvalue);
        Consumer<Date> printer = date -> {
            System.out.println(date);
            System.out.flush();
        };
        // CompletableFuture computation
        CompletableFuture.runAsync(task)
                         .thenApply((v) -> longValue.get())
                         .thenApply(dateConverter)
                         .thenAccept(printer);
}

Methods thenRunhave thenApplyversions . thenAccept_ AsyncThis means that these stages will be executed in a new thread. It will be taken from a special pool, so it is not known in advance what kind of flow it will be, new or old. It all depends on how difficult the tasks are. In addition to these methods, there are three more interesting possibilities. For clarity, let’s imagine that we have a certain service that receives a message from somewhere and it takes time:

public static class NewsService {
	public static String getMessage() {
		try {
			Thread.currentThread().sleep(3000);
			return "Message";
		} catch (InterruptedException e) {
			throw new IllegalStateException(e);
		}
	}
}

Now, let's look at the other features that CompletableFuture. We can combine the result CompletableFuturewith the result of another CompletableFuture:

Supplier newsSupplier = () -> NewsService.getMessage();

CompletableFuture<String> reader = CompletableFuture.supplyAsync(newsSupplier);
CompletableFuture.completedFuture("!!")
				 .thenCombine(reader, (a, b) -> b + a)
				 .thenAccept(result -> System.out.println(result))
				 .get();

It’s worth noting that by default the threads will be daemon threads, so for clarity get, we use to wait for the result. And we can not only combine (combine), but also return CompletableFuture:

CompletableFuture.completedFuture(2L)
				.thenCompose((val) -> CompletableFuture.completedFuture(val + 2))
                               .thenAccept(result -> System.out.println(result));

Here I would like to note that for brevity, the method was used CompletableFuture.completedFuture. This method does not create a new thread, so the rest of the chain will be executed in the same thread in which it was called completedFuture. There is also a method thenAcceptBoth. It is very similar to accept, but if thenAcceptit accepts consumer, then thenAcceptBothit accepts another CompletableStage+ as input BiConsumer, that is consumer, which accepts 2 sources as input, not one. There is another interesting possibility with the word Either: These methods accept an alternative CompletableStageand will be executed on the one CompletableStagethat is executed first. And I would like to finish this review with another interesting feature CompletableFuture- error handling.

CompletableFuture.completedFuture(2L)
				 .thenApply((a) -> {
					throw new IllegalStateException("error");
				 }).thenApply((a) -> 3L)
				 //.exceptionally(ex -> 0L)
				 .thenAccept(val -> System.out.println(val));

This code will not do anything, because... an exception will be thrown and nothing will happen. But if we uncomment exceptionally, then we define the behavior. CompletableFutureI also recommend watching the following video on this topic :

• CompletableFuture. I want to take it and use it (2015)
• CompletableFuture in Java 8, asynchronous processing done right

In my humble opinion, these videos are some of the most visual on the Internet. It should be clear from them how it all works, what arsenal we have and why it’s all needed.

Conclusion

Hopefully it's now clear how threads can be used to retrieve calculations after they've been calculated. Additional material:

• Writing asynchronous code with CompletableFuture
• Introduction to CompletionStage and CompletableFuture
• Java 8 CompletableFuture. Part 2
• Guide To Completable Future

#Viacheslav