Java run thread async

Parallel and Asynchronous Programming in Java 8

Parallel programming no longer needs to be an insurmountable obstacle in the hunt for faster code.

Parallel code, which is code that runs on more than one thread, was once the nightmare of many an experienced developer, but Java 8 brought a lot of changes that should make this performance-boosting trick a lot more manageable.

Parallel Streams

Before Java 8 there was a big difference between parallel (or concurrent) code and sequential code. It was also very hard to debug non-sequential code. Simply setting a breakpoint and going through the flow like you would normally do, would remove the parallel aspect, which is a problem if that is what is causing the bug.

Luckily, Java 8 gave us streams, the greatest thing for Java developers since the bean. If you don’t know what they are, the Stream API makes it possible to handle sequences of elements in a functional matter. (Check our comparison between streams and .NET’s LINQ here.) One of the advantages of streams is that the structure of the code stays the same: whether it’s sequential or concurrent, it stays just as readable.

To make your code run parallel, you simply use .parallelStream() instead of .stream(), (or stream.parallel(), if you are not the creator of the stream).

The speed question

Parallel code gets its speed benefit from using multiple threads instead of the single one that sequential code uses. Deciding how many threads to create can be a tricky question because more threads don’t always result in faster code: if you use too many threads the performance of your code might actually go down.

There are a couple of rules that will tell you what number of threads to choose. This depends mostly on the kind of operation that you want to perform and the number of available cores.

Computation intensive operations should use a number of threads lower than or equal to the number of cores, while IO intensive operations like copying files have no use for the CPU and can therefore use a higher number of threads. The code doesn’t know which case is applicable unless you tell it what to do. Otherwise, it will default to a number of threads equal to the number of cores.

There are two main cases when it can be useful to run your code parallel instead of sequential: time-consuming tasks and tasks run on big collections. Java 8 brought a new way of handling those big collections, namely with streams. Streams have built-in efficiency by laziness: they use lazy evaluation which saves resources by not doing more than necessary. This is not the same as parallelism, which doesn’t care about the resources as long as it goes faster. So for big collections, you probably don’t need classic parallelism.

Going Async

Lessons from JavaScript

It is a rare occurrence that a Java developer can say that they learned something from looking at JavaScript, but when it comes to asynchronous programming, JavaScript actually got it right first. As a fundamentally async language, JavaScript has a lot of experience with how painful it can be when badly implemented. It started with callbacks and was later replaced by promises. An important benefit of promises is that it has two “channels”: one for data and one for errors. A JavaScript promise might look something like this:

func .then(f1) .catch(e1) .then(f2) .catch(e2);

So when the original function has a successful result, f1 is called, but if an error was thrown e1 will be called. This might bring it back to the successful track (f2), or result in another error (e2). You can go from data track to error track and back.

The Java version of JavaScript promises is called CompletableFuture.

CompletableFuture

CompletableFuture implements both the Future and the CompletionStage interface. Future already existed pre-Java8, but it wasn’t very developer-friendly by itself. You could only get the result of the asynchronous computation by using the .get() method, which blocked the rest (making the async part pretty pointless most of the time) and you needed to implement each possible scenario manually. Adding the CompletionStage interface was the breakthrough that made asynchronous programming in Java workable.

CompletionStage is a promise, namely the promise that the computation will eventually be done. It contains a bunch of methods that let you attach callbacks that will be executed on that completion. Now we can handle the result without blocking.

There are two main methods that let you start the asynchronous part of your code: supplyAsync if you want to do something with the result of the method, and runAsync if you don’t.

CompletableFuture.runAsync(() → System.out.println("Run async in completable future " + Thread.currentThread())); CompletableFuture.supplyAsync(() → 5);

Now you can add those callbacks to handle the result of your supplyAsync

CompletableFuture.supplyAsync(() → 5) .thenApply(i → i * 3) .thenAccept(i → System.out.println(“The result is “ + i) .thenRun(() → System.out.println("Finished."));

.thenApply is similar to the .map function for streams: it performs a transformation. In the example above it takes the result (5) and multiplies it by 3. It will then pass that result (15) further down the pipe.

.thenAccept performs a method on the result without transforming it. It will also not return a result. Here it will print “The result is 15” to the console. It can be compared to the .foreach method for streams.

.thenRun doesn’t use the result of the async operation and also doesn’t return anything, it just waits to call its Runnable until the previous step is completed.

Asyncing your async

All of the above callback methods also come in an async version: thenRunAsync, thenApplyAsync, etc. These versions can run on their own thread and they give you extra control because you can tell it which ForkJoinPool to use.

If you don’t use the async version, then the callbacks will all be executed on the same thread.

When things go wrong

When something goes wrong, the exceptionally method is used to handle the exception. You can give it a method that returns a value to get back on the data track, or throw a (new) exception.

… .exceptionally(ex → new Foo()) .thenAccept(this::bar);

Combine and Compose

You can chain multiple CompletableFutures by using the thenCompose method. Without it, the result would be nested CompletableFutures. This makes thenCompose and thenApply like flatMap and map for streams.

CompletableFuture.supplyAsync(() -> "Hello") .thenCompose(s -> CompletableFuture .supplyAsync(() -> s + "World"));

If you want to combine the result of two CompletableFutures, you will need a method conveniently called thenCombine.

future.thenCombine(future2, Integer::sum) .thenAccept(value → System.out.println(value));

As you can see in the example above, the result of the callback in thenCombine can be handled like a normal CompletableFuture with all your favourite CompletionStage methods.

Conclusion

Parallel programming no longer needs to be an insurmountable obstacle in the hunt for faster code. Java 8 makes the process as straightforward as can be, so that any piece of code that could possibly benefit from it, can be pulled, kicking and screaming on all threads, into the multi-core future that is, in fact, just the present day. By which I mean: it’s easy to do, so give it a try and see its advantages for yourself.

Dit blog is geschreven door de specialisten van Foreach.

Inmiddels is Foreach onderdeel van iO. Meer weten? Neem gerust contact op!

Источник

Async Programming in Java: Part I

Join the DZone community and get the full member experience.

As a backend engineer, we face situations to process the data asynchronously. Today let’s see how it’s done in java and various way to do it.

Starting from Thread, Runnable, Callable, Future (and its extended ScheduledFuture), CompletableFuture, and of course, ExecutorService and ForkJoinPool. We will see all of them, but one by one.

Thread

The very basic yet so powerful component of Java concurrency is Thread . The Thread of Java is actually associated with the Thread of the Operating System. The very basic way to create a Thread is by extending it and overriding the run method:

public class TestThread extends Thread

TestThread t = new TestThread();

t.start();// starting the thread, causes the run method be called

Starting the thread causes the run() method to be called.
You may ask; yes, Thread has tons of other methods that can be overridden:

• In most cases, we don’t want to override other methods of the thread.
• Once we extend the Thread class, the extending class losses its ability to extend further as Java does not support multiple inheritances.
• Each thread has its own object when we extend it, and it’s not good for memory health when there are tons of Objects of the extended Thread created.

Java addresses these issues with the Runnable interface. In fact, Thread has an overloaded method that takes Runnable.

Runnable

Runnable is an interface that has only one method: run() . Yes, Runnable is a functional interface, and its instance can be created with the lambda function. Yet it’s an easy way to do this; for complex things, we might like to implement it. See the difference here. It’s all about the requirement:

 Runnable runnable = ()->System.out.println("I'm a runnable from lambda.");

// With implemention, we can hold the data and related stuff that we want to process.

// Otherwise we got to manage them in the launching thread

public class RunnableImplemented implements Runnable

 ListObject> mayBeAListOfData;

 public RunnableImplemented(ListObject> mayBeAListOfData, 

 Object mayBeAService, Object mayBeADao) 

 this.mayBeAListOfData = mayBeAListOfData;

 this.mayBeAService = mayBeAService;

Though Runnable has a run() method, it’s not a Thread but just a Java class until it’s taken control by (passed to) Thread. The starting of the thread causes the runnable object’s run() method to be called.

 private static Runnable runnable = ()->System.out.println("I'm a runnable from lambda.");

 public static void main(String[] args) 

 Thread t = new Thread(runnable);// takes runnable here

Cool, we learnt how to create threads with Thread as well as Runnable. But have you noticed Thread (or Runnable either),

• The Runnable method doesn’t return anything?
• Also doesn’t have exception handling. You have to surround your code that throws an Exception with try and catch block

Yep, Java got it solved in version 1.5, and it’s Callable.

Callable

Callable is a generic interface. Why? The type of the return value as the generic type. Callable is too a functional interface and call() is the only method, a no-argument method that throws Exception and returns generic type value.

The implementing Callable is very similar to Runnable:

private static CallableInteger> callable = ()-> 

 String data = "I'm in callable.";

public static void main(String[] args) 

 ExecutorService executor = Executors.newSingleThreadExecutor();

 FutureInteger> callFuture = executor.submit(callable);

 Integer integer = callFuture.get();

See here, the call method processes the data and returns a value that can be collected post-execution. But is there a huge difference in invoking it? We use ExecutorService to invoke and Future to hold the result. Let’s talk about why.

As you can see, there is no controlled behavior of creating and running the Threads (Runnable or Callable too). We may want to control the number of threads running at a time as each of them associated with OS’s threads. The number of Threads we run should be lesser than the number of available CPU cores. All together, Java solves it by ExecutorService interface.

I think it’s really a lot for today. Maybe in Part II, we will discuss Executors and different types of ExecutorService, and the Future interface.

Opinions expressed by DZone contributors are their own.

Источник

Running a Task Asynchronously in Java With CompletableFuture

Here’s how to use CompletableFuture to run a task asynchronously in Java.

import java.util.concurrent.CompletableFuture; import java.util.concurrent.ForkJoinPool; import java.util.logging.Level; import java.util.logging.Logger; import static java.util.concurrent.TimeUnit.SECONDS; public class AsyncTask < private static final Logger LOGGER = Logger.getLogger(AsyncTask.class.getName()); public static void main(String[] args) < CompletableFuture.runAsync(AsyncTask::task); LOGGER.info("Message from the main thread. Note that this message is logged before the async task ends."); LOGGER.info("Waiting for the async task to end."); boolean isQuiecent = ForkJoinPool.commonPool().awaitQuiescence(5, SECONDS); if (isQuiecent) < LOGGER.info("Async task ended."); >else < LOGGER.log(Level.SEVERE, "The async task is taking too long to finish. This program will end anyway."); >> private static void task() < LOGGER.info("Async task starting. This message is logged by the async task thread"); try < Thread.sleep(1000); LOGGER.info("Async task is ending. This message is logged by the async task thread"); >catch (InterruptedException e) < Thread.currentThread().interrupt(); LOGGER.log(Level.SEVERE, "The async task thread was interrupted.", e); >> >

If you have any questions, leave a comment or ask me on my social media.

Источник