Multithreaded programming in Java 8. Part I

Welcome to the first part of the guidelines for parallel programming in Java 8. In this section we look at simple examples of how to execute code in parallel by a flow of tasks and services of performers.

For the first time Concurrency API was introduced with the release of Java 5, and since then has evolved with each version of Java. Most of the examples can be implemented in older versions, but in this article I'm going to use lambda expressions. If you are still not familiar with the innovations of Java 8, I recommend to see myguide.

Streams and objectives

All modern operating systems support parallel execution of code using the processes andflows. The process - an instance of a program that runs independently of the others. For example, when you run a program in Java, the operating system creates a new process that runs parallel to the others. Within the processes we use flows, thereby to squeeze out the maximum processor capacity.

Streams (threads) in Java are supported starting with JDK 1.0. Before you start a thread, it is necessary to provide a piece of code, which is usually called a "challenge» (task). This is done by implementing the interfaceRunnable, which has only one method with no arguments, returns void -  run (). Here is an example of how this works:

1 2 3 4 5 6 7 8 9 10 11

Runnable task = () -> {     String threadName = Thread.currentThread().getName();     System.out.println("Hello " + threadName); };   task.run();   Thread thread = new Thread(task); thread.start();   System.out.println("Done!");

Because the interface Runnable function, we can use lambda expressions, which appeared in Java 8. In the example we create a task that displays the name of the current stream to the console, and run it first on the main thread, and then - in a separate.

The result of this code might look like this:

1 2 3

Hello main Hello Thread-0 Done!

or so:

1 2 3

Hello main Done! Hello Thread-0

Due to the parallel execution, we can not say to our current run before or after the withdrawal of «Done!» On the screen. This feature makes parallel programming difficult in larger applications.

Threads can be suspended for some time. This is very useful if we want to simulate a long-running task. For example:

1 2 3 4 5 6 7 8 9 10 11 12 13 14

Runnable runnable = () -> {     try {         String name = Thread.currentThread().getName();         System.out.println("Foo " + name);         TimeUnit.SECONDS.sleep(1);         System.out.println("Bar " + name);     }     catch (InterruptedException e) {         e.printStackTrace();     } };   Thread thread = new Thread(runnable); thread.start();

When you run this code, you will see a second delay between the output of the first and second line on the screen. TimeUnit - a useful class for working with time units, but the same can be done with Thread.sleep (1000).

Working with streams directly uncomfortable and fraught with errors. Therefore, in 2004, we added in Java 5 Concurrency API. It is in the package java.util.concurrent contains many useful classes and methods for multi-threaded programming. Since Concurrency API is continuously developing and evolving.

Let us now take a closer look one of the most important parts of Concurrency API - Service performers  (executor services).

Artists

Concurrency API introduces the concept of service of the Executive (ExecutorService) - a high-level replacement with threads directly. Artists perform tasks asynchronously and usually use the thread pool, so we do not have to create them manually. Everything flows from the pool will be used again after the task, and therefore, we can create problems in the application as much as we want, using one performer.

That will look like our first example of using the artist:

1 2 3 4 5 6 7

ExecutorService executor = Executors.newSingleThreadExecutor(); executor.submit(() -> {     String threadName = Thread.currentThread().getName();     System.out.println("Hello " + threadName); });   // => Hello pool-1-thread-1

Class Executors provides convenient factory methods to create a variety of services artists. In this case, we used a performer with a single thread.

The result looks just like last time. But this code is an important difference - it never stops. The work necessary to complete the performers clearly. For this interface ExecutorService there are two methods: shutdown (), which waits for the running tasks and shutdownNow (), which stops the artist immediately.

That is how I prefer to stop the singers:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

try {     System.out.println("attempt to shutdown executor");     executor.shutdown();     executor.awaitTermination(5, TimeUnit.SECONDS); } catch (InterruptedException e) {     System.err.println("tasks interrupted"); } finally {     if (!executor.isTerminated()) {         System.err.println("cancel non-finished tasks");     }     executor.shutdownNow();     System.out.println("shutdown finished"); }

Artist tries to shut down, waiting for the completion of the running tasks for a certain period of time (5 seconds). After this time he stopped interrupting all pending tasks.

Callable and Future

Also Runnable, performers may take a different kind of problems, called Callable. Callable - it is also functional interface but, unlike Runnable, it can return a value.

Let's write a task that returns an integer after a moment:

1 2 3 4 5 6 7 8 9

Callable task = () -> {     try {         TimeUnit.SECONDS.sleep(1);         return 123;     }     catch (InterruptedException e) {         throw new IllegalStateException("task interrupted", e);     } };

Callable-tasks can also be transferred to the performers. But how to get the result that they return? Since the method submit () does not wait for completion of the task, the Executive can not return a result of the problem directly. Instead, the artist returns a special object Future, which we will be able to request the result of the task.

1 2 3 4 5 6 7 8 9

ExecutorService executor = Executors.newFixedThreadPool(1); Future<Integer> future = executor.submit(task);   System.out.println("future done? " + future.isDone());   Integer result = future.get();   System.out.println("future done? " + future.isDone()); System.out.print("result: " + result);

After sending the task executor we first check whether its execution is completed, the methodisDone (). Since the problem has a delay of one second before you return the number, I am more than sure that it is not yet completed.

Calling get () blocks the thread, and waits for the task, and then return the result of its implementation. Now future.isDone () returns true, and we will see in the console as follows:

1 2 3

future done? false future done? true result: 123

Tasks are rigidly connected with the service of artists, and if you stop it, try to get the result of problem will throw an exception:

1 2	executor.shutdownNow(); future.get();

You may have noticed that this time we create a service a little differently: by the methodnewFixedThreadPool (1), which will return the artist with a pool in one stream. This is equivalent to calling newSingleThreadExecutor (), but we can change the number of threads in the pool.

Timeouts

Any method call future.get () blocks the thread until then, until the task is completed. In the worst case, the task will not end ever, blocking your application. This can be avoided by passing the timeout:

1 2 3 4 5 6 7 8 9 10 11 12 13

ExecutorService executor = Executors.newFixedThreadPool(1);   Future<Integer> future = executor.submit(() -> {     try {         TimeUnit.SECONDS.sleep(2);         return 123;     }     catch (InterruptedException e) {         throw new IllegalStateException("task interrupted", e);     } });   future.get(1, TimeUnit.SECONDS);

Running this code will cause TimeoutException:

1 2	Exception in thread "main" java.util.concurrent.TimeoutException     at java.util.concurrent.FutureTask.get(FutureTask.java:205)

You have probably already guessed, why this exception was thrown: we specify the maximum waiting time of the task in one second, while its execution takes two.

InvokeAll

Artists can make a list of tasks to perform using the method invokeAll (), which takes a collection of callable-tasks, and returns a list of the Future.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

ExecutorService executor = Executors.newWorkStealingPool();   List<Callable<String>> callables = Arrays.asList(         () -> "task1",         () -> "task2",         () -> "task3");   executor.invokeAll(callables)     .stream()     .map(future -> {         try {             return future.get();         }         catch (Exception e) {             throw new IllegalStateException(e);         }     })     .forEach(System.out::println);

In this example, we used a functional Java threads 8 to handle tasks that were returned byinvokeAll. We walked all the problems and brought them to the result on the console. If you are not familiar with Java thread 8, see myguide.

InvokeAny

Another way to give back to perform several tasks - method invokeAny (). It works a little differently: instead of returning Future it blocks the flow before the end at least one task, and returns the result.

To show how this method works, create a method that emulates the behavior of various tasks.He will return Callable, which will return the specified string after the necessary delay:

1 2 3 4 5 6

Callable callable(String result, long sleepSeconds) {     return () -> {         TimeUnit.SECONDS.sleep(sleepSeconds);         return result;     }; }

Use this method to create multiple tasks with different lines and delays from one to three seconds. Sending these tasks through a method executor invokeAny () will return the result of problems with the least delay. In this case «task2»:

1 2 3 4 5 6 7 8 9 10 11

ExecutorService executor = Executors.newWorkStealingPool();   List<Callable<String>> callables = Arrays.asList(     callable("task1", 2),     callable("task2", 1),     callable("task3", 3));   String result = executor.invokeAny(callables); System.out.println(result);   // => task2

In the above example we used another type of artists who created a methodnewWorkStealingPool () . This method appeared in Java 8, and does not behave like the others: instead of using a fixed number of threads it creates ForkJoinPool a certain parallelism(parallelism size), by default equal to the number of cores of the machine.

ForkJoinPool first introduced in Java 7, and we will review it in more detail in the following sections of our management. Now let's look at the performers with a scheduler (scheduledexecutors).

Artists with a scheduler

We already know how to give the task executor and get the result. In order to periodically run the task, we can use the thread pool to a scheduler.

ScheduledExecutorService able to run the task once or several times at a specified interval.

This example shows how to get the singer to perform a task after three seconds:

1 2 3 4 5 6 7 8 9

ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);   Runnable task = () -> System.out.println("Scheduling: " + System.nanoTime()); ScheduledFuture<?>; future = executor.schedule(task, 3, TimeUnit.SECONDS);   TimeUnit.MILLISECONDS.sleep(1337);   long remainingDelay = future.getDelay(TimeUnit.MILLISECONDS); System.out.printf("Remaining Delay: %sms", remainingDelay);

When we pass the task scheduler, it returns a special type of Future -ScheduledFuture, which provides a method getDelay () for the remaining time before the launch.

Performers with scheduler there are two methods to set the task: scheduleAtFixedRate () andscheduleWithFixedDelay (). The first set problem with a certain interval, e.g., one second:

1 2 3 4 5 6 7

ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);   Runnable task = () -> System.out.println("Scheduling: " + System.nanoTime());   int initialDelay = 0; int period = 1; executor.scheduleAtFixedRate(task, initialDelay, period, TimeUnit.SECONDS);

In addition, it takes an initial delay, which specifies the time until the first start-up.

Please note that the method scheduleAtFixedRate () does not take into account the execution of the task. So, if you put a task that is performed for two seconds, at intervals of one, the thread pool will sooner or later overflow.

In this case, you must use the method scheduleWithFixedDelay (). It works much the same as the previous, but this interval will be counted from the time of completion of the previous task.

1 2 3 4 5 6 7 8 9 10 11 12 13

ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);   Runnable task = () -> {     try {         TimeUnit.SECONDS.sleep(2);         System.out.println("Scheduling: " + System.nanoTime());     }     catch (InterruptedException e) {         System.err.println("task interrupted");     } };   executor.scheduleWithFixedDelay(task, 0, 1, TimeUnit.SECONDS);

In this example we set the task with a delay of one second between the end of the task and the start of the next. The initial delay is not, and each task is performed for two seconds. Thus, the task will start at 0, 3, 6, 9, and so on. G. Second. As you can see, the methodscheduleWithFixedDelay () is very useful when we can not tell in advance how the task will run.

It was the first part of a series of articles about multi-threaded programming. We strongly recommend to dismantle the above examples alone. All are available on GitHub. Feel free forkan repository and add it to the favorites.

I hope you enjoyed the article. If you have any questions, you can ask them in a tweet.