Multithreading in Java and Spring Boot

Published by jamie on

1. Creating Threads in Java

1.1 Extending the Thread Class

public class MyThread extends Thread {
    public void run() {
        // Code to be executed in the new thread
        for (int i = 0; i < 5; i++) {
            System.out.println("Thread " + Thread.currentThread().getId() + ": " + i);
        }
    }
}

public class Main {
    public static void main(String[] args) {
        MyThread thread1 = new MyThread();
        MyThread thread2 = new MyThread();
        thread1.start(); // Starts a new thread
        thread2.start(); // Starts another thread
    }
}

1.2 Implementing the Runnable Interface

public class MyRunnable implements Runnable {
    public void run() {
        // Code to be executed in the new thread
        for (int i = 0; i < 5; i++) {
            System.out.println("Thread " + Thread.currentThread().getId() + ": " + i);
        }
    }
}

public class Main {
    public static void main(String[] args) {
        Thread thread1 = new Thread(new MyRunnable());
        Thread thread2 = new Thread(new MyRunnable());
        thread1.start(); // Starts a new thread
        thread2.start(); // Starts another thread
    }
}

1.3 Implementing the Callable Interface

This method is similar to the method of implementing the Runnable interface, except that the call() method of the Callable interface can return a result and throw an exception. You need to create a class that implements the Callable interface and override the call() method to define the task to be performed by the thread in the call() method. Then, create an object of this class and pass it as a parameter to the submit() method of the ExecutorService class, and finally call the get() method of the Future class to obtain the return result of the call() method.

package com.pany.camp.thread;

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

public class MyCallable implements Callable<Integer> {
    private int num;

    public MyCallable(int num) {
        this.num = num;
    }

    @Override
    public Integer call() throws Exception {
        int sum = 0;
        for (int i = 1; i <= num; i++) {
            sum += i;
        }
        return sum;
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        MyCallable myCallable = new MyCallable(100);
        FutureTask<Integer> futureTask = new FutureTask<>(myCallable);
        Thread thread = new Thread(futureTask);
        thread.start();
        int result = futureTask.get();
        System.out.println("sum from 1 to 100：" + result);
    }
}

1.4 Using Lambda Expressions (Java 8+)

public class Main {
    public static void main(String[] args) {
        Runnable task = () -> {
            for (int i = 0; i < 5; i++) {
                System.out.println("Thread " + Thread.currentThread().getId() + ": " + i);
            }
        };

        Thread thread1 = new Thread(task);
        Thread thread2 = new Thread(task);
        thread1.start();
        thread2.start();
    }
}

1.5 CompletableFuture

CompletableFuture is a powerful asynchronous programming tool introduced in Java 8. It supports chain calls, multiple asynchronous task combinations, exception handling and other features, making asynchronous programming more flexible.

import java.util.concurrent.CompletableFuture;

public class CompletableFutureExample {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return "Task completed!";
        });

        future.thenAccept(result -> System.out.println(result));

        try {
            future.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

1.5.1 runAsync()

Perform asynchronous calculations (no return value)

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // your code
})

1.5.2 supplyAsync()

Perform asynchronous calculations and return results.

1.5.3 thenApply()

The thenApply method is used to transform the result of a CompletableFuture. It accepts a Function parameter that is applied to the result after the asynchronous task is completed and returns a new CompletableFuture containing the transformed result.

CompletableFuture<String> originalFuture = CompletableFuture.supplyAsync(() -> "Hello");

CompletableFuture<String> transformedFuture = originalFuture.thenApply(s -> s.toUpperCase());

1.5.4 thenCompose()

The thenCompose method is used to connect two asynchronous tasks and merge their results into a new CompletableFuture. It accepts a Function parameter that returns a new CompletableFuture.

CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> secondFuture = CompletableFuture.supplyAsync(() -> " World");

CompletableFuture<String> combinedFuture = firstFuture.thenCompose(s -> secondFuture.thenApply(t -> s + t));

1.5.5 thenCombine()

The thenCombine method is used to combine the results of two independent CompletableFuture. It accepts two parameters: the first is another CompletableFuture, and the second is a BiFunction used to merge the two results.

CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> secondFuture = CompletableFuture.supplyAsync(() -> " World");

CompletableFuture<String> combinedFuture = firstFuture.thenCombine(secondFuture, (s, t) -> s + t);

1.5.6 exceptionally()

The exceptionally method is used to handle exceptional situations. It accepts a Function parameter, which is executed when an exception occurs in an asynchronous task and returns a replacement value or another asynchronous task.

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 模拟异常
    throw new RuntimeException("Error");
});

CompletableFuture<String> result = future.exceptionally(ex -> "Handled Exception");

1.5.7 handle()

The handle method is similar to exceptionally, but it can handle both normal results and exceptions. It accepts a BiFunction parameter, which is executed when the asynchronous task is completed and can handle normal results or exceptions.

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");

CompletableFuture<String> result = future.handle((res, ex) -> {
    if (ex != null) {
        return "Handled Exception";
    } else {
        return res.toUpperCase();
    }
});

1.5.8 allOf()

The allOf method accepts an array of CompletableFuture and returns a new CompletableFuture that will be completed when all tasks in the array are completed.

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");

CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(future1, future2);

1.5.9 anyOf()

The anyOf method also accepts an array of CompletableFuture s, but the returned CompletableFuture completes when any of the tasks in the array completes.

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");

CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(future1, future2);

1.5.10 Summary

Application scenarios:

  • FutureTask: Suitable for simple asynchronous task execution, such as when using ExecutorService for thread pool management.
  • CompletableFuture: Suitable for complex asynchronous programming requirements, especially scenarios that require combining multiple asynchronous tasks, such as parallel operations, serial operations, etc.

Pros and Conds:

  • FutureTask

    • advantage:

      • Simple and easy to use, suitable for basic asynchronous task execution.

      • Relatively lightweight and does not introduce too much complexity.

    • shortcoming:
      • It does not support the combination of multiple asynchronous tasks and cannot conveniently perform concurrent operations.
  • CompletableFuture

    • advantage:

      • Powerful asynchronous programming features support the combination and coordinated execution of multiple asynchronous tasks.

      • Rich exception handling mechanisms make error handling more flexible.

    • shortcoming:
      • Has a steeper learning curve and can be more complex to use than simple asynchronous tasks.

1.6 Thread Pool

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorExample {
    public static void main(String[] args) {
        // Create a thread pool with 2 threads
        ExecutorService executor = Executors.newFixedThreadPool(2);

        for (int i = 0; i < 5; i++) {
            executor.submit(() -> {
                System.out.println("Thread " + Thread.currentThread().getId() + " is running");
            });
        }

        // Shutdown the executor service
        executor.shutdown();
    }
}

2. Multithreading in Spring and Spring Boot

2.1 Implementing Asynchronous Methods with @Async

Spring’s @Async annotation is a simple yet powerful way to run methods asynchronously. When applied to a method, it allows the method to execute in a separate thread, freeing up the main thread to continue executing other tasks. This is particularly useful for long-running operations that you don’t want to block the main flow of your application.

Enabling Asynchronous Execution: To use the @Async annotation, you must first enable async support in your Spring Boot application by adding the @EnableAsync annotation to a configuration class.

import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;

@Configuration
@EnableAsync
public class AsyncConfig {
}

Once you’ve enabled asynchronous support, you can apply the @Async annotation to any method in a Spring-managed bean.

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.util.concurrent.CompletableFuture;

@Service
public class AsyncService {

    @Async
    public CompletableFuture<String> performAsyncTask() {
        try {
            // Simulate a long-running task
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return CompletableFuture.completedFuture("Task completed");
    }
}

2.2 Customizing Async Behavior with ThreadPoolTaskExecutor

By default, Spring Boot uses a simple thread pool when executing asynchronous tasks. However, in a production environment, you may want to customize the thread pool to suit your application’s needs. Spring Boot provides the ThreadPoolTaskExecutor for this purpose, which allows you to configure thread pool parameters such as core pool size, max pool size, and queue capacity.

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.annotation.EnableAsync;

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "taskExecutor")
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(500);
        executor.setThreadNamePrefix("MyExecutor-");
        executor.initialize();
        return executor;
    }
}

In this example, the ThreadPoolTaskExecutor is configured with:

  • Core pool size: Minimum number of threads that will always be available.

  • Max pool size: Maximum number of threads that can be created.

  • Queue capacity: How many tasks can be queued before new threads are created.

  • Thread name prefix: Custom naming pattern for threads in this executor.

To use this custom thread pool, the method annotated with @Async will automatically utilize the taskExecutor defined in the configuration.

General Guidelines:

  • Core Pool Size: The number of threads that are always available (even if idle). A good default is equal to the number of CPU cores for CPU-bound tasks.

  • Max Pool Size: The maximum number of threads the pool can create. Increase this if tasks are often blocked or waiting on external resources.

  • Queue Capacity: The number of tasks that can be queued before new threads are created. For short tasks, a larger queue is efficient. For long tasks, consider a smaller queue to avoid long delays.

2.3 Scheduling Tasks with @Scheduled

Another powerful feature in Spring Boot is the ability to schedule tasks to run at specific intervals or times using the @Scheduled annotation. This is useful for running background jobs, such as database cleanup, report generation, or sending periodic notifications.

Enabling Scheduling: To use the @Scheduled annotation, you must enable scheduling by adding @EnableScheduling to a configuration class.

import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;

@Configuration
@EnableScheduling
public class SchedulerConfig {
}

Now, you can create methods that are scheduled to run at fixed intervals, using cron expressions, or after a specific delay.

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class ScheduledTask {

    @Scheduled(fixedRate = 5000)
    public void performTask() {
        System.out.println("Task executed every 5 seconds");
    }

    @Scheduled(cron = "0 0 12 * * ?")
    public void performTaskAtNoon() {
        System.out.println("Task executed at 12 PM every day");
    }
}

In this example, the performTask() method runs every 5 seconds, while performTaskAtNoon() runs every day at 12 PM. Spring handles the scheduling internally, ensuring that the tasks are executed on time without blocking the main application thread.

2.4 TaskExecutor for Higher-Level Abstraction

Spring provides the TaskExecutor interface as a higher-level abstraction for managing asynchronous tasks. This allows you to execute Runnable tasks without dealing with the lower-level details of thread management.

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Service;

@Service
public class TaskExecutorService {

    @Autowired
    private TaskExecutor taskExecutor;

    public void executeAsyncTask() {
        taskExecutor.execute(() -> {
            System.out.println("Task executed asynchronously using TaskExecutor");
        });
    }
}

Here, the TaskExecutor interface is injected, and the executeAsyncTask() method uses it to execute a task asynchronously. This provides a simple and clean way to run background tasks without needing to create new threads manually.

For more advanced use cases, Spring Boot also supports reactive programming with Spring WebFlux. This is particularly useful for non-blocking, event-driven applications that need to handle a large number of concurrent requests without consuming excessive system resources.

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import java.time.Duration;

@RestController
public class ReactiveController {

    @GetMapping("/reactive")
    public Flux<String> getReactiveStream() {
        return Flux.just("Message 1", "Message 2", "Message 3")
                   .delayElements(Duration.ofSeconds(1));
    }
}

In this example, a Flux (a reactive stream) is returned, which emits a series of messages with a delay of 1 second between each. This non-blocking model ensures that the server can handle other requests while the stream is being processed.

3. Thread Safety in Spring Boot

When dealing with multithreading in Spring Boot, thread safety is a critical consideration, especially since Spring beans are typically singleton by default. This means that multiple threads could access the same instance of a bean simultaneously, potentially leading to data corruption or inconsistent states if shared resources are not handled correctly

3.1 Common Thread Safety Issues:

  1. Race Conditions: Occurs when two or more threads access shared data and try to change it at the same time.

  2. Deadlocks: Happen when two or more threads are blocked forever, waiting for each other to release resources.

  3. Memory Consistency Errors: Result when multiple threads access shared variables without proper synchronization.

3.2 Handling Thread Safety in Spring Boot:

  1. Use of synchronized Keyword: You can synchronize methods or code blocks to ensure that only one thread can execute them at a time, preventing race conditions.

 @Service
 public class CounterService {
     private int count = 0;

     public synchronized void increment() {
         count++;
     }

     public synchronized int getCount() {
         return count;
     }
 }

In this example, both increment() and getCount() are synchronized to ensure that the counter is safely updated and retrieved, preventing multiple threads from corrupting the shared variable.

2. Atomic Variables: For lightweight thread-safe operations, you can use Java’s Atomic classes (e.g., AtomicInteger, AtomicBoolean), which provide methods for thread-safe manipulation of variables without the overhead of synchronization.

 import java.util.concurrent.atomic.AtomicInteger;

 @Service
 public class AtomicCounterService {
     private AtomicInteger count = new AtomicInteger(0);

     public void increment() {
         count.incrementAndGet();
     }

     public int getCount() {
         return count.get();
     }
 }

3. ThreadLocal Variables: ThreadLocal variables allow you to store data that is specific to the current thread, ensuring that each thread has its own copy of the variable. This is useful for managing thread-specific data like request contexts.

 public class UserContext {
     private static final ThreadLocal<String> userContext = new ThreadLocal<>();

     public static void setUser(String user) {
         userContext.set(user);
     }

     public static String getUser() {
         return userContext.get();
     }

     public static void clear() {
         userContext.remove();
     }
 }
Categories: JavaSpringboot
Tags:

0 Comments

Leave a Reply

Avatar placeholder

Your email address will not be published. Required fields are marked *

Related Posts

hibernate

Hibernate in Spring Boot – Quick Start

Set up dependency <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-jpa</artifactId> </dependency> Configuring JPA and Hibernate  in application.properties file spring.datasource.url=jdbc:mysql://localhost:3306/dbname spring.datasource.username=root spring.datasource.password=password spring.jpa.hibernate.ddl-auto=update or application.yml spring: application: name: demoapp datasource: url: jdbc:mysql://localhost:3306/dbname username: root password: password driver-class-name: com.mysql.cj.jdbc.Driver jpa: hibernate: Read more…

Backend Development

Spring Interceptor

Function Can constitute a stack of interceptors to complete specific functions, such as logging, login judgment, permission check and so on. Benefits Interceptors also allow you to modularize generic code as reusable classes. Applications of Read more…

Backend Development

Java 8 Optional Usage

Introduction An Optional is a container object which may or may not contain a non-null value. It was introduced in Java 8 to cure the curse of NullPointerExceptions. In essence Optional is a wrapper class Read more…

Table of Contents

← Catalogue