Xử lý bất đồng bộ Java

Trước khi xem xét Spring @Async, khái niệm về đồng bộ, bất đồng bộ và đa luồng là điều cần thiết. Giả sử bạn đã biết các khái niệm đó, hãy cùng tìm hiểu cách xử lý bất đồng bộ Java thuần túy thông qua mã. Nếu bạn quen thuộc với luồng Java, bạn có thể bỏ qua chương này.

public class MessageService {

    public void print(String message) {
        System.out.println(message);
    }
}

public class Main {

    public static void main(String[] args) {
        MessageService messageService = new MessageService();

        for (int i = 1; i <= 100; i++) {
            messageService.print(i + "");
        }
    }

Nếu bạn tạo một chức năng đơn giản để nhận và xuất thông báo theo cách đồng bộ, bạn có thể viết mã như trên. Thay đổi nó thành phương thức bất đồng bộ đa luồng, bạn có thể viết mã nguồn như sau.

public class MessageService {

    public void print(String message) {
        new Thread(() -> System.out.println(message))
                .start();
    }
}

public class Main {

    public static void main(String[] args) {
        MessageService messageService = new MessageService();

        for (int i = 1; i <= 100; i++) {
            messageService.print(i + "");
        }
    }

Tuy nhiên, phương pháp này rất nguy hiểm vì bạn không thể quản lý Luồng. Ví dụ, nếu có 10.000 cuộc gọi cùng lúc, bạn sẽ phải tạo 10.000 Luồng trong thời gian rất ngắn. Chi phí tạo Luồng không hề nhỏ, vì vậy nó sẽ ảnh hưởng đến hiệu suất của chương trình, thậm chí có thể gây ra lỗi OOM. Do đó, để quản lý Luồng, bạn cần triển khai Pool Luồng, và Java cung cấp lớp ExecutorService.

public class MessageService {

    private final ExecutorService executorService = Executors.newFixedThreadPool(10);

    public void print(String message) {
        executorService.submit(() -> System.out.println(message));
    }
}

public class Main {

    public static void main(String[] args) {
        MessageService messageService = new MessageService();

        for (int i = 1; i <= 100; i++) {
            messageService.print(i + "");
        }
    }

Số lượng luồng tổng thể được giới hạn ở 10 và chúng ta đã có thể xử lý xử lý bất đồng bộ theo cách đa luồng mà chúng ta muốn. Tuy nhiên, phương thức này yêu cầu bạn phải áp dụng phương thức submit() của ExecutorService cho mỗi phương thức mà bạn muốn xử lý bất đồng bộ, vì vậy bạn cần phải thực hiện công việc sửa đổi lặp đi lặp lại. Nói cách khác, nếu bạn muốn thay đổi một phương thức được viết ban đầu theo logic đồng bộ thành bất đồng bộ, bạn phải thay đổi logic của chính phương thức đó.

Spring @Async

Phương pháp đơn giản

@EnableAsync
@SpringBootApplication
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }

}

@Service
public class MessageService {

    @Async
    public void print(String message) {
        System.out.println(message);
    }
}

@RequiredArgsConstructor
@RestController
public class MessageController {

    private final MessageService messageService;

    @GetMapping("/messages")
    @ResponseStatus(code = HttpStatus.OK)
    public void printMessage() {
        for (int i = 1; i <= 100; i++) {
            messageService.print(i + "");
        }
    }

Chỉ cần thêm chú thích @EnableAsync vào lớp Application và thêm chú thích @Async vào phương thức logic đồng bộ mà bạn muốn xử lý bất đồng bộ. Tuy nhiên, phương pháp này có vấn đề là không quản lý luồng. Bởi vì cấu hình mặc định của @Async là sử dụng SimpleAsyncTaskExecutor, đây không phải là pool luồng mà chỉ đơn giản là vai trò tạo ra luồng.

Cách sử dụng pool luồng

Đầu tiên, hãy loại bỏ @EnableAsync khỏi lớp Application. Nếu lớp Application của bạn có cấu hình @EnableAutoConfiguration hoặc @SpringBootApplication, nó sẽ đọc thông tin bean threadPoolTaskExecutor trong lớp SpringAsyncConfig được cấu hình (sẽ được tạo ở bên dưới) trong thời gian chạy.

@Configuration
@EnableAsync // Phải thêm vào lớp Async Config, không phải Application.
public class SpringAsyncConfig {

    @Bean(name = "threadPoolTaskExecutor")
    public Executor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(3); // Số luồng cơ bản
        taskExecutor.setMaxPoolSize(30); // Số lượng luồng tối đa
        taskExecutor.setQueueCapacity(100); // Kích thước hàng đợi
        taskExecutor.setThreadNamePrefix("Executor-");
        return taskExecutor;
    }

Bạn có thể đặt kích thước core và max. Ở đây, bạn có thể mong đợi rằng nó sẽ hoạt động với kích thước core ban đầu và sau đó tăng số lượng luồng lên kích thước max nếu không thể xử lý thêm công việc, nhưng không phải vậy.

Nó tạo ra một LinkedBlockingQueue có kích thước Integer.MAX_VALUE bên trong và đặt các công việc vào Hàng đợi nếu không thể xử lý các công việc trong số lượng luồng cơ bản. Khi Hàng đợi đầy, nó sẽ tạo ra nhiều luồng nhất định để xử lý các công việc.

Nếu bạn cảm thấy khó chịu khi đặt kích thước Hàng đợi thành Integer.MAX_VALUE, bạn có thể đặt queueCapacity. Nếu bạn đặt nó như trên, nó sẽ xử lý với 3 luồng ban đầu, đặt công việc vào Hàng đợi có kích thước 100 nếu tốc độ xử lý chậm lại, và tạo tối đa 30 luồng để xử lý công việc nếu có thêm yêu cầu.

Sau khi cấu hình pool luồng hoàn tất, bạn chỉ cần thêm tên bean vào phương thức có chú thích @Async.

@Service
public class MessageService {

    @Async("threadPoolTaskExecutor")
    public void print(String message) {
        System.out.println(message);
    }

Nếu bạn muốn đặt nhiều loại pool luồng, hãy tạo nhiều phương thức tạo bean như threadPoolTaskExecutor() và chèn bean pool luồng mong muốn khi đặt cấu hình @Async.

Kiểu trả về theo kiểu trả về

Không có giá trị trả về

Đây là trường hợp phương thức cần xử lý bất đồng bộ không cần trả lại kết quả xử lý. Trong trường hợp này, bạn có thể đặt kiểu trả về của chú thích @Async thành void.

Có giá trị trả về

Bạn có thể sử dụng kiểu Future, ListenableFuture, CompletableFuture làm kiểu trả về. Hình thức trả về của phương thức bất đồng bộ được gói trong new AsyncResult().

[Future]

@Service
public class MessageService {

    @Async
    public Future print(String message) throws InterruptedException {
        System.out.println("Task Start - " + message);
        Thread.sleep(3000);
        return new AsyncResult<>("jayon-" + message);
    }
}

@RequiredArgsConstructor
@RestController
public class MessageController {

    private final MessageService messageService;

    @GetMapping("/messages")
    @ResponseStatus(code = HttpStatus.OK)
    public void printMessage() throws ExecutionException, InterruptedException {
        for (int i = 1; i <= 5; i++) {
            Future future = messageService.print(i + "");
            System.out.println(future.get());
        }
    }
}

// Kết quả thực thi
Task Start - 1
jayon-1
Task Start - 2
jayon-2
Task Start - 3
jayon-3
Task Start - 4
jayon-4
Task Start - 5

future.get() đóng vai trò chờ đợi cho đến khi kết quả yêu cầu đến thông qua chặn. Vì vậy, nó trở thành phương thức chặn bất đồng bộ và hiệu suất không tốt. Thông thường, Future không được sử dụng.

[ListenableFuture]

@Service
public class MessageService {

    @Async
    public ListenableFuture print(String message) throws InterruptedException {
        System.out.println("Task Start - " + message);
        Thread.sleep(3000);
        return new AsyncResult<>("jayon-" + message);
    }
}

@RequiredArgsConstructor
@RestController
public class MessageController {

    private final MessageService messageService;

    @GetMapping("/messages")
    @ResponseStatus(code = HttpStatus.OK)
    public void printMessage() throws InterruptedException {
        for (int i = 1; i <= 5; i++) {
            ListenableFuture listenableFuture = messageService.print(i + "");
            listenableFuture.addCallback(System.out::println, error -> System.out.println(error.getMessage()));
        }
    }
}

// Kết quả thực thi
Task Start - 1
Task Start - 3
Task Start - 2
jayon-1
jayon-2
Task Start - 5
jayon-3
Task Start - 4
jayon-4

ListenableFuture cho phép bạn xử lý công việc theo cách không chặn thông qua hồi gọi. Tham số đầu tiên của phương thức addCallback() là phương thức hồi gọi hoàn thành công việc, tham số thứ hai là phương thức hồi gọi lỗi. Lưu ý rằng vì số lượng luồng cơ bản của pool luồng được đặt thành 3, nên bạn có thể thấy thông báo “Bắt đầu nhiệm vụ” được in ra đầu tiên.

[CompletableFuture]

ListenableFuture cũng có thể triển khai logic không chặn, nhưng nếu cần thêm hồi gọi vào trong hồi gọi, nó sẽ dẫn đến mã rất phức tạp được gọi là địa ngục hồi gọi.

Tất nhiên, thời gian này chúng ta sẽ không đề cập đến CompletableFuture một cách chi tiết, vì vậy nếu bạn muốn biết thêm về mã xử lý hồi gọi phức tạp, hãy tham khảo liên kết nguồn ở bên dưới.

@Service
public class MessageService {

    @Async
    public CompletableFuture print(String message) throws InterruptedException {
        System.out.println("Task Start - " + message);
        Thread.sleep(3000);
        return new AsyncResult<>("jayon-" + message).completable();
    }
}

@RequiredArgsConstructor
@RestController
public class MessageController {

    private final MessageService messageService;

    @GetMapping("/messages")
    @ResponseStatus(code = HttpStatus.OK)
    public void printMessage() throws InterruptedException {
        for (int i = 1; i <= 5; i++) {
            CompletableFuture completableFuture = messageService.print(i + "");
            completableFuture
                    .thenAccept(System.out::println)
                    .exceptionally(error -> {
                        System.out.println(error.getMessage());
                        return null;
                    });
        }
    }

Nó có tính khả đọc tốt hơn so với việc xác định hồi gọi của ListenableFuture và thực hiện chức năng không chặn một cách hoàn hảo. Do đó, khi sử dụng @Async, nếu cần giá trị trả về, bạn nên sử dụng CompletableFuture.

Ưu điểm của @Async

Nhà phát triển có thể viết phương thức theo cách đồng bộ và chỉ cần thêm chú thích @Async vào trên phương thức nếu họ muốn sử dụng phương thức bất đồng bộ. Vì vậy, bạn có thể tạo mã có khả năng bảo trì tốt cho đồng bộ và bất đồng bộ.

Lưu ý về @Async

Để sử dụng chức năng @Async, bạn cần khai báo chú thích @EnableAsync, và nếu không có bất kỳ cấu hình nào khác, nó sẽ hoạt động ở chế độ proxy. Nói cách khác, tất cả các phương thức bất đồng bộ hoạt động với chú thích @Async đều tuân theo các hạn chế của Spring AOP. Để biết thêm lý do, hãy tham khảobài đăng này..

• Thêm @Async vào phương thức private sẽ không làm cho AOP hoạt động.
• AOP sẽ không hoạt động khi gọi lẫn nhau giữa các phương thức trong cùng một đối tượng.

Nguồn

• http://dveamer.github.io/java/SpringAsync.html
• https://brunch.co.kr/@springboot/401
• https://kapentaz.github.io/spring/Spring-ThreadPoolTaskExecutor-설정/#
• https://www.hungrydiver.co.kr/bbs/detail/develop?id=2