Source: https://dzone.com/articles/happens-before-in-java-or-how-to-write-thread-safe

Happens-Before In Java Or How To Write a Thread-Safe Application

This article explains the notion of happens-before in Java, including ways to install it, what guarantee it gives, advantages it brings, and how to use it.

by Dmitry Egorov CORE · Jul. 26, 22 · Java Zone · Tutorial

Multithreading is the most complex part of Java. Happens-before is a relation that gives a guarantee of allowing the writing of predictable code in multithreading a reality. Such code is also known as thread-safe code. Unfortunately, Oracle Java documentation about this notion is hard to read. So in this article, I'll mainly explain what happens-before is in human language and provide detailed examples.

Happens-Before Solves the Main Multithreading Problem

Before we start learning the happens-before notion, we have to understand the reason for creating it. Once multithreading comes to the scene, your code might become inconsistent because shared objects between threads might have different and unpredictable values. Let's review a simple example. In that example, we will update values in one thread and read and print them in another.

Expected Behavior (False Expectations)

Now let's think about what values we might see in the second thread while printing.  Depending on progress in the first thread, we might expect 3 situations: 

1. X and y are not initialized, so 0 and 0 will be printed.
2. X is set and y is not set yet; then, 1 and 0 will be printed.
3. X and y are set; then, 1 and 1 will be printed.

 Now let's run the program a million times and see what values have been printed. 

Among a million prints, there were only 3 different options printed (with my hardware, Windows OS, and my JDK). For some reason, the B case never appeared because it's not as confusing as an 0,1 pair, which looks impossible. The only thing we have to learn here is that results are unpredictable. 

Why Shared Memory Might Be Inconsistent

This is a pretty big topic and it's not the main subject of this article, so I will just briefly introduce the most known reasons.  

Reordering or JVM Optimizations

JVM can change the order of the instructions which won't make any significant changes in the program. For example, it might change the order of variable initialization.

CPU Memory Cache

Each thread can be executed in a separated CPU. CPU has its own cache with different variables copied. When one thread updates the value in the CPU, another thread will still have an outdated value. We can illustrate it like this:

Both threads read data from shared RAM memory after processing its data with resynchronization. By default, nothing forces the CPU to update the shared RAM memory value.

What Is Actually Important: Thread-Safe Code

Everything explained earlier is only valuable for passing interviews; but in practice, all developers need is a way to predictably and consistently code in a multithreading reality, also known as thread-safe code. Fortunately, Java offers ways to write such code.

Happens-Before Gives a Guarantee of Visibility of Same Fields in Different Threads

The happens-before relationship is described in Oracle Documentation, Chapter 17.4.5. If between two threads, we installed the happens-before relation, then the second (ending thread) will see all changes that happened in the first thread. Therefore, the second thread execution will be similar to a single thread application.

Install Happens-Before Relationship: Making Fields Visible (Using a Volatile Keyword)

The first way to install the happens-before relation is to mark a shared variable with a volatile keyword. If a variable is volatile, then the happens-before relation will be installed between every write and subsequent reads.

A write to a volatile field (§8.3.1.4) happens-before every subsequent read of that field.

This means that:

• Once Y is updated in the first thread, it will be visible in the second thread.
• When Y became visible in the second thread, all fields are set before Y.

Considering this, we have to make the next changes in order to have x and y consistent:

• Make Y volatile.
• Make X read after Y in the second thread.
• Check that Y is set to 1.

Considering all the changes, we rewrite our code in this way next:

This might be a bit complicated. One thing you have to pay attention to is that the blue portal means that the happens-before relation is installed (starting point) and the yellow portal means that the happens-before relation is completed (and all data "passed" through is received with the latest, up-to-date state).

Install Happens-Before Relationship: Making Ordered Access to Fields (Using Synchronized Monitor)

Java provides a way to organize ordered access to the instructions known as a Java monitor. I have described what it is and how it works in a set of three articles:

• Multithreading Java and Interviews Part 1: An Introduction
• Multithreading Java and Interviews Part 2: Mutex, the Java Monitor Model
• Multithreading Java and Interviews Part 3: Wait and Notify All

This subject is pretty complex. I also recommend reading these articles:

• What is a Monitor in Computer Science?
• Guide to the Synchronized Keyword in Java

The happens-before relationship is installed between any lock acquire and lock release. So, if we rewrite our code using locks, we also have a happens-before guarantee:

As you can see, both fields are not volatile; but when the first thread updates the value and releases the lock, then in the second thread when the lock is acquired, the value also gets updated due to the happened-before relation. This rule applied to all locks from java.util.concurrent that use a Java monitor under the hood.  

Install Happens-Before Relationship: Thread Start and Join

This might be the easiest case compared to the previous two. Java specification says:  

- A call to start() on a thread happens-before any actions in the started thread.

- All actions in a thread happen-before any other thread successfully returns from a join() on that thread.

When the thread starts, it sees the correct state of shared variables of the parent thread. This is backward when the parent thread is released after join(): it receives the latest state of the child thread.  

Let's rewrite our example and illustrate how bean data is passed to the child thread when we call the start() method:

The example explains that values set in the main thread are successfully passed to the child thread due to the installation of the happens-before relation, and thanks to start() method.

Now let's illustrate the same logic, but when data is passed from the children to parent on join() method:

In this example, we pass data from the child thread to the main awaiting thread. Once a child thread ends its run and releases join, then values are passed correctly due to the existence of the happens-before relationship.

Are Examples Production-Ready and Recommended To Use?

No, no, no! The examples that I provided served only one purpose: to explain what happens-before is, and how it works. In practice, you can't be too careful in a multithreading environment. To better avoid things I have shown: 

• In the first case with visibility (volatile keyword), I marked only one field as volatile. The code will work as expected; but in practice, you can change the order of reading and the "broken" happens-before relationship. 
• In the case of the synchronization monitor, I didn't mark fields as volatile because it's not necessary. Don't do this. Keep fields volatile even if they are changed inside the lock.

Also, it is worth mentioning that in the first example (with volatile), I didn't mention when what happens-before relation is not enough when you take care of consistency. Volatile keywords give you safe reads, but not simultaneous writes (leads to race conditions). This subject is outside of this article.

 Source: https://dzone.com/articles/multi-threading-in-spring-boot-using-completablefu

Multi-Threading in Spring Boot Using CompletableFuture

Learn more about multi-threading in Spring Boot Using CompletableFuture.

by

Swathi Prasad Sep. 17, 19 · Java Zone · Tutorial

Multi-threading is similar to multi-tasking, but it enables the processing of executing multiple threads simultaneously, rather than multiple processes. CompletableFuture, which was introduced in Java 8, provides an easy way to write asynchronous, non-blocking, and multi-threaded code.

The Future interface was introduced in Java 5 to handle asynchronous computations. But, this interface did not have any methods to combine multiple asynchronous computations and handle all the possible errors. The  CompletableFuture implements Future interface, it can combine multiple asynchronous computations, handle possible errors and offers much more capabilities.

Let's get down to writing some code and see the benefits.

Create a sample Spring Boot project and add the following dependencies.

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <groupId>com.techshard.future</groupId>

    <artifactId>springboot-future</artifactId>

    <version>1.0-SNAPSHOT</version>

    <parent>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter-parent</artifactId>

        <version>2.1.8.RELEASE</version>

        <relativePath />

    </parent>

    <properties>

        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

    </properties>

    <dependencies>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-web</artifactId>

        </dependency>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-data-jpa</artifactId>

        </dependency>

        <dependency>

            <groupId>com.h2database</groupId>

            <artifactId>h2</artifactId>

            <scope>runtime</scope>

        </dependency>

        <dependency>

            <groupId>org.slf4j</groupId>

            <artifactId>slf4j-api</artifactId>

        </dependency>

        <dependency>

            <groupId>org.projectlombok</groupId>

            <artifactId>lombok</artifactId>

            <version>1.18.10</version>

            <optional>true</optional>

        </dependency>

    </dependencies>

    <build>

        <plugins>

            <plugin>

                <groupId>org.springframework.boot</groupId>

                <artifactId>spring-boot-maven-plugin</artifactId>

            </plugin>

        </plugins>

    </build>

</project>

In this article, we will be using sample data about cars. We will create a JPA entity Car and a corresponding JPA repository.

package com.techshard.future.dao.entity;

import lombok.Data;

import lombok.EqualsAndHashCode;

import javax.persistence.*;

import javax.validation.constraints.NotNull;

import java.io.Serializable;

@Data

@EqualsAndHashCode

@Entity

public class Car implements Serializable {

    private static final long serialVersionUID = 1L;

    @Id

    @Column (name = "ID", nullable = false)

    @GeneratedValue (strategy = GenerationType.IDENTITY)

    private long id;

    @NotNull

    @Column(nullable=false)

    private String manufacturer;

    @NotNull

    @Column(nullable=false)

    private String model;

    @NotNull

    @Column(nullable=false)

    private String type;

}

package com.techshard.future.dao.repository;

import com.techshard.future.dao.entity.Car;

import org.springframework.data.jpa.repository.JpaRepository;

import org.springframework.stereotype.Repository;

@Repository

public interface CarRepository extends JpaRepository<Car, Long> {

}

Let us now create a configuration class that will be used to enable and configure the asynchronous method execution.

package com.techshard.future;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.scheduling.annotation.EnableAsync;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;

@Configuration

@EnableAsync

public class AsyncConfiguration {

    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncConfiguration.class);

    @Bean (name = "taskExecutor")

    public Executor taskExecutor() {

        LOGGER.debug("Creating Async Task Executor");

        final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

        executor.setCorePoolSize(2);

        executor.setMaxPoolSize(2);

        executor.setQueueCapacity(100);

        executor.setThreadNamePrefix("CarThread-");

        executor.initialize();

        return executor;

    }

}

The @EnableAsync annotation enables Spring's ability to run  @Async methods in a background thread pool. The bean taskExecutor helps to customize the thread executor such as configuring the number of threads for an application, queue limit size, and so on. Spring will specifically look for this bean when the server is started. If this bean is not defined, Spring will create SimpleAsyncTaskExecutor by default.

We will now create a service and @Async methods.

package com.techshard.future.service;

import com.techshard.future.dao.entity.Car;

import com.techshard.future.dao.repository.CarRepository;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.scheduling.annotation.Async;

import org.springframework.stereotype.Service;

import org.springframework.web.multipart.MultipartFile;

import java.io.*;

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.CompletableFuture;

@Service

public class CarService {

    private static final Logger LOGGER = LoggerFactory.getLogger(CarService.class);

    @Autowired

    private CarRepository carRepository;

    @Async

    public CompletableFuture<List<Car>> saveCars(final MultipartFile file) throws Exception {

        final long start = System.currentTimeMillis();

        List<Car> cars = parseCSVFile(file);

        LOGGER.info("Saving a list of cars of size {} records", cars.size());

        cars = carRepository.saveAll(cars);

        LOGGER.info("Elapsed time: {}", (System.currentTimeMillis() - start));

        return CompletableFuture.completedFuture(cars);

    }

    private List<Car> parseCSVFile(final MultipartFile file) throws Exception {

        final List<Car> cars=new ArrayList<>();

        try {

            try (final BufferedReader br = new BufferedReader(new InputStreamReader(file.getInputStream()))) {

                String line;

                while ((line=br.readLine()) != null) {

                    final String[] data=line.split(";");

                    final Car car=new Car();

                    car.setManufacturer(data[0]);

                    car.setModel(data[1]);

                    car.setType(data[2]);

                    cars.add(car);

                }

                return cars;

            }

        } catch(final IOException e) {

            LOGGER.error("Failed to parse CSV file {}", e);

            throw new Exception("Failed to parse CSV file {}", e);

        }

    }

    @Async

    public CompletableFuture<List<Car>> getAllCars() {

        LOGGER.info("Request to get a list of cars");

        final List<Car> cars = carRepository.findAll();

        return CompletableFuture.completedFuture(cars);

    }

}

Here, we have two @Async methods: saveCar() and  getAllCars(). The first one accepts a multipart file, parses it, and stores the data in the database. The second method reads the data from the database.

Both methods are returning a new CompletableFuture that was already completed with the given values.

Let us create a Rest Controller and provide some endpoints:

package com.techshard.future.controller;

import com.techshard.future.dao.entity.Car;

import com.techshard.future.service.CarService;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.http.HttpStatus;

import org.springframework.http.MediaType;

import org.springframework.http.ResponseEntity;

import org.springframework.web.bind.annotation.*;

import org.springframework.web.multipart.MultipartFile;

import java.io.File;

import java.util.List;

import java.util.concurrent.CompletableFuture;

import java.util.function.Function;

@RestController

@RequestMapping("/api/car")

public class CarController {

    private static final Logger LOGGER = LoggerFactory.getLogger(CarController.class);

    @Autowired

    private CarService carService;

    @RequestMapping (method = RequestMethod.POST, consumes={MediaType.MULTIPART_FORM_DATA_VALUE},

            produces={MediaType.APPLICATION_JSON_VALUE})

    public @ResponseBody ResponseEntity uploadFile(

            @RequestParam (value = "files") MultipartFile[] files) {

        try {

            for(final MultipartFile file: files) {

                carService.saveCars(file);

            }

            return ResponseEntity.status(HttpStatus.CREATED).build();

        } catch(final Exception e) {

            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();

        }

    }

    @RequestMapping (method = RequestMethod.GET, consumes={MediaType.APPLICATION_JSON_VALUE},

            produces={MediaType.APPLICATION_JSON_VALUE})

    public @ResponseBody CompletableFuture<ResponseEntity> getAllCars() {

        return carService.getAllCars().<ResponseEntity>thenApply(ResponseEntity::ok)

                .exceptionally(handleGetCarFailure);

    }

    private static Function<Throwable, ResponseEntity<? extends List<Car>>> handleGetCarFailure = throwable -> {

        LOGGER.error("Failed to read records: {}", throwable);

        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();

    };

}

The first REST endpoint accepts a list of multipart files. The second endpoint is to read the data. As you notice the GET endpoint, there is some difference in the return statement. We are returning a list of cars and we are also handling exceptions.

The function  handleGetCarFailure is invoked when the CompletableFuture completes exceptionally, otherwise, if this CompletableFuture completes normally, it returns a list of cars to the client.

Testing the Application

Run the Spring Boot Application. Once the server is started, test the POST endpoint. The sample screenshot from Postman tool.

Make sure to provide Content-Type as multipart\form-data in the headers section. When you send a request, you will notice that two threads have started at the same time, one thread for each file.

Let us now test the GET endpoint.

Now, just modify the GET endpoint as follows:

@RequestMapping (method = RequestMethod.GET, consumes={MediaType.APPLICATION_JSON_VALUE},

            produces={MediaType.APPLICATION_JSON_VALUE})

    public @ResponseBody ResponseEntity getAllCars() {

        try {

            CompletableFuture<List<Car>> cars1=carService.getAllCars();

            CompletableFuture<List<Car>> cars2=carService.getAllCars();

            CompletableFuture<List<Car>> cars3=carService.getAllCars();

            CompletableFuture.allOf(cars1, cars2, cars3).join();

            return ResponseEntity.status(HttpStatus.OK).build();

        } catch(final Exception e) {

            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();

        }

    }

Here, we are calling the Async method 3 times. The  CompletableFuture.allOf() will wait until all the CompletableFutures have been completed, and join() will combine the results. Note that this is just for demonstration purposes.

Add Thread.sleep(1000L) in  getAllCars() of the  CarService class. We are giving a delay of 1 second just for testing purpose.

Restart the application and test GET endpoint again.

As you see in the above screenshot, the first two calls to the Async method have started simultaneously. The third call has started with a delay of 1 second.

Remember that we have configured only 2 threads that can be used simultaneously. When at least one of the two threads becomes free, the third request to the Async method will be made.

Conclusion

In this article, we've seen some typical use cases of the  CompletableFuture. Let me know if you have any comments or suggestions in the comments section below.

The source code for this article can be found on this GitHub repository.