การประมวลผลแบบอะซิงโครนัสของ Java

ก่อนที่จะดู Spring @Async เราจำเป็นต้องเข้าใจแนวคิดเกี่ยวกับการซิงโครนัส อะซิงโครนัส และการมัลติเธรด สมมติว่าคุณเข้าใจแนวคิดเหล่านี้แล้ว ลองมาดูวิธีการประมวลผลแบบอะซิงโครนัสของ Java แบบธรรมดาผ่านโค้ด ถ้าคุณคุ้นเคยกับเธรดของ Java คุณสามารถข้ามบทนี้ไปได้

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 + "");
        }
    }

ถ้าคุณสร้างฟังก์ชันเพื่อรับ message และแสดงผล output ออกมาแบบซิงโครนัส คุณสามารถเขียนโค้ดได้ดังนี้ ถ้าคุณเปลี่ยนให้เป็นแบบมัลติเธรดแบบอะซิงโครนัส คุณสามารถเขียนโค้ดได้ดังนี้

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 + "");
        }
    }

อย่างไรก็ตาม วิธีนี้ไม่สามารถจัดการ Thread ได้ จึงมีความเสี่ยงสูง ตัวอย่างเช่น ถ้ามีการเรียกใช้ 10,000 ครั้งพร้อมกัน คุณจะต้องสร้าง Thread 10,000 ตัวในเวลาอันสั้น ค่าใช้จ่ายในการสร้าง Thread ไม่ใช่เรื่องเล็กน้อย ดังนั้นจึงส่งผลเสียต่อประสิทธิภาพของโปรแกรม และอาจเกิดข้อผิดพลาด OOM ขึ้นได้ ดังนั้น คุณจำเป็นต้องใช้ Thread Pool เพื่อจัดการ Thread และ Java มี ExecutorService Class

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 + "");
        }
    }

เราได้กำหนดจำนวน Thread ทั้งหมดให้เป็น 10 และสามารถทำการประมวลผลแบบอะซิงโครนัสแบบมัลติเธรดได้อย่างถูกต้อง อย่างไรก็ตาม วิธีนี้ต้องใช้เมธอด submit() ของ ExecutorService ในทุกเมธอดที่ต้องการประมวลผลแบบอะซิงโครนัส ซึ่งหมายความว่าคุณต้องแก้ไขโค้ดซ้ำๆ นั่นคือ ถ้าคุณต้องการเปลี่ยนเมธอดที่เขียนด้วยตรรกะแบบซิงโครนัสเป็นแบบอะซิงโครนัส คุณจำเป็นต้องเปลี่ยนตรรกะของเมธอดนั้นเอง

Spring @Async

วิธีง่ายๆ

@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 + "");
        }
    }

เพียงแค่ใส่ @EnableAsync Annotation ไว้บน Application Class และ @Async Annotation ไว้บนเมธอดที่มีตรรกะแบบซิงโครนัส ที่คุณต้องการประมวลผลแบบอะซิงโครนัส แต่ปัญหาของวิธีนี้คือมันไม่จัดการ Thread เพราะการตั้งค่าเริ่มต้นของ @Async คือการใช้ SimpleAsyncTaskExecutor ซึ่งไม่ใช่ Thread Pool แต่เป็นเพียงการสร้าง Thread

วิธีการใช้ Thread Pool

ก่อนอื่น ให้ลบ @EnableAsync ออกจาก Application Class ถ้า Application Class มีการตั้งค่า @EnableAutoConfiguration หรือ @SpringBootApplication ไว้ ณ เวลาทำงาน Spring จะอ่านข้อมูล threadPoolTaskExecutor bean (ซึ่งเราจะสร้างขึ้นในภายหลัง) จาก SpringAsyncConfig Class (ซึ่งถูกกำหนดโดย @Configuration)

@Configuration
@EnableAsync // ต้องใส่ใน Async Setting Class ไม่ใช่ Application
public class SpringAsyncConfig {

    @Bean(name = "threadPoolTaskExecutor")
    public Executor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(3); // จำนวน Thread เริ่มต้น
        taskExecutor.setMaxPoolSize(30); // จำนวน Thread สูงสุด
        taskExecutor.setQueueCapacity(100); // ขนาดของ Queue
        taskExecutor.setThreadNamePrefix("Executor-");
        return taskExecutor;
    }

คุณสามารถตั้งค่าขนาด Core และ Max ได้ คุณอาจคิดว่าในกรณีนี้จะทำงานด้วยขนาด Core เริ่มแรกและถ้า Thread ไม่สามารถ ประมวลผลงานได้อีก จะเพิ่มขนาด Thread เป็นขนาด Max แต่ไม่ใช่เช่นนั้น

ภายในจะสร้าง LinkedBlockingQueue ที่มีขนาด Integer.MAX_VALUE และจะใส่งานเข้าไปใน Queue ถ้า Thread ขนาด Core ไม่สามารถประมวลผลงานได้ ถ้า Queue เต็ม จะสร้าง Thread ขนาด Max เพื่อประมวลผล

ถ้าคุณรู้สึกว่าการตั้งค่าขนาด Queue เป็น Integer.MAX_VALUE มีภาระมาก คุณสามารถตั้งค่า queueCapacity ได้ ถ้าตั้งค่าดังนี้ ตอนแรกจะมี Thread 3 ตัวในการประมวลผล ถ้าความเร็วในการประมวลผลลดลง งานจะถูกใส่ใน Queue ขนาด 100 และถ้ามีการร้องขอเข้ามาเพิ่มอีก จะสร้าง Thread สูงสุด 30 ตัวเพื่อประมวลผล

หลังจากตั้งค่า Thread Pool เรียบร้อยแล้ว คุณสามารถใส่ชื่อ bean ไว้ในเมธอดที่มี @Async Annotation

@Service
public class MessageService {

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

ถ้าคุณต้องการตั้งค่าชนิดของ Thread Pool หลายประเภท คุณสามารถสร้างเมธอด bean creation หลายเมธอด เช่น threadPoolTaskExecutor() และใส่ bean ของ Thread Pool ที่ต้องการใน @Async setting

รูปแบบของค่าที่ส่งกลับตามชนิดของค่าส่งกลับ

กรณีที่ไม่มีค่าส่งกลับ

กรณีที่เมธอดที่ต้องการประมวลผลแบบอะซิงโครนัสไม่จำเป็นต้องส่งผลลัพธ์กลับ คุณสามารถตั้งค่าชนิดของค่าส่งกลับของ @Async Annotation เป็น void

กรณีที่มีค่าส่งกลับ

คุณสามารถใช้ชนิดของค่าส่งกลับเป็น Future, ListenableFuture, CompletableFuture คุณสามารถใช้ 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());
        }
    }
}

// ผลลัพธ์การดำเนินการ
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() จะรอจนกระทั่งผลลัพธ์ของการร้องขอถูกส่งกลับมาโดยใช้การบล็อก ดังนั้นจึงเป็นการบล็อกแบบอะซิงโครนัส ซึ่งทำให้ประสิทธิภาพไม่ดี โดยทั่วไปไม่นิยมใช้ Future

[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()));
        }
    }
}

// ผลลัพธ์การดำเนินการ
Task Start - 1
Task Start - 3
Task Start - 2
jayon-1
jayon-2
Task Start - 5
jayon-3
Task Start - 4
jayon-4

ListenableFuture สามารถประมวลผลงานแบบ non-blocking ผ่าน callback พารามิเตอร์แรกของเมธอด addCallback() คือ callback เมธอดสำหรับการทำงานเสร็จสมบูรณ์ พารามิเตอร์ที่สองคือ callback เมธอดสำหรับการทำงานล้มเหลว หมายเหตุ: เนื่องจากเรากำหนดขนาด Core ของ Thread Pool ไว้ที่ 3 เราจึงเห็นว่าข้อความ “Task Start” แสดง ครั้งแรก 3 ครั้ง

[CompletableFuture]

ถึงแม้ ListenableFuture จะสามารถใช้สร้างตรรกะแบบ non-blocking ได้ แต่ถ้าต้องใช้ callback ใน callback อาจทำให้เกิดโค้ดที่ซับซ้อนซึ่งเรียกว่า callback hell

แน่นอน บทเรียนนี้ไม่ได้เจาะลึกเรื่อง CompletableFuture ดังนั้น ถ้าคุณอยากรู้ว่าโค้ดที่จัดการกับ callback ที่ซับซ้อนเป็นอย่างไร คุณสามารถดูลิงก์แหล่งข้อมูลด้านล่าง

@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;
                    });
        }
    }

โค้ดอ่านง่ายกว่าการกำหนด callback ของ ListenableFuture และทำงานแบบ non-blocking อย่างสมบูรณ์ ดังนั้น ถ้าคุณต้องการส่งค่ากลับเมื่อใช้ @Async แนะนำให้ใช้ CompletableFuture

ข้อดีของ @Async

นักพัฒนาสามารถเขียนเมธอดแบบซิงโครนัส และถ้าต้องการแบบอะซิงโครนัส ก็แค่ใส่ @Async Annotation บนเมธอดนั้น จึงทำให้คุณสามารถสร้างโค้ดที่ง่ายต่อการดูแลรักษาสำหรับการซิงโครนัสและอะซิงโครนัส

ข้อควรระวังของ @Async

คุณต้องประกาศ @EnableAsync Annotation เพื่อใช้ฟังก์ชัน @Async แต่ถ้าไม่มีการตั้งค่าเพิ่มเติม จะทำงานในโหมด Proxy หมายความว่าเมธอดแบบอะซิงโครนัสที่ใช้ @Async Annotation จะต้องปฏิบัติตามข้อจำกัดของ Spring AOP คุณสามารถดูสาเหตุได้จาก โพสต์นี้โปรดดูลิงก์

• AOP จะไม่ทำงานถ้าใส่ @Async บนเมธอดแบบ private
• AOP จะไม่ทำงานถ้าเรียกใช้เมธอดภายในวัตถุเดียวกัน

แหล่งข้อมูล

• 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