Giới thiệu về CompletableFuture trong Java

Trong bài viết trước, mình đã giới thiệu với các bạn về đối tượng Future trong Java và mình cũng đã nói với các bạn rằng khi sử dụng phương thức get() của đối tượng Future, chương trình của chúng ta sẽ bị block cho đến khi tất cả các tác vụ hoàn thành. Để giải quyết vấn đề block chương trình này, từ phiên bản 8, Java giới thiệu đối tượng CompletableFuture giúp chúng ta giải quyết vấn đề này. Cụ thể như thế nào? Hãy cùng tìm hiểu với mình trong bài viết này nhé!

Để dễ so sánh giữa đối tượng Future và đối tượng CompletableFuture, mình sẽ làm ví dụ với đối tượng Future trước.

Đối tượng Callable của mình sẽ có nội dung như sau:

package com.huongdanjava.javaexample;

import java.util.concurrent.Callable;

public class Calculator implements Callable<Integer> {
    
    private int a;
    private int b;

    public Calculator(int a, int b) {
        this.a = a;
        this.b = b;
    }
    
    public Integer call() throws Exception {
        Thread.sleep(3000);
        
        return a + b;
    }
}

package com.huongdanjava.javaexample;

import java.util.concurrent.Callable;

public class Calculator implements Callable<Integer> {

private int a;

private int b;

public Calculator(int a, int b) {

this.a = a;

this.b = b;

}

public Integer call() throws Exception {

Thread.sleep(3000);

return a + b;

}

Các bạn để ý là trong phương thức call() của đối tượng Callable, mình đang sleep 3s để demo cho việc xử lý lâu của Thread.

Bây giờ mình sẽ viết một main class sử dụng đối tượng Future. Code của class này như sau:

package com.huongdanjava.javaexample;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class Example {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // Create thread pool using Executor Framework
        ExecutorService executor = Executors.newFixedThreadPool(10);

        // Create new Calculator object
        Calculator c = new Calculator(2, 3);

        Future<Integer> f = executor.submit(c);

        System.out.println(f.get());

        System.out.println("End...");

        executor.shutdown();
    }
}

package com.huongdanjava.javaexample;

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

public class Example {

public static void main(String[] args) throws ExecutionException, InterruptedException {

// Create thread pool using Executor Framework

ExecutorService executor = Executors.newFixedThreadPool(10);

// Create new Calculator object

Calculator c = new Calculator(2, 3);

Future<Integer> f = executor.submit(c);

System.out.println(f.get());

System.out.println("End...");

executor.shutdown();

}

Khi chạy ví dụ này, các bạn sẽ thấy sau 3s chúng ta sẽ nhận được kết quả rồi mới in kết quả ra được và dòng “End..” cũng mới được output:

Bây giờ, mình sẽ sử dụng đối tượng CompletableFuture để chương trình của chúng ta không bị block như vậy nữa và cũng nhận được kết quả như khi sử dụng với đối tượng Future.

Đầu tiên, chúng ta cần phải submit đối tượng Calculator vào một Thread khác trước. CompletableFuture có nhiều methods khác nhau giúp chúng ta làm việc này. Trong bài viết này, mình sẽ sử dụng phương thức supplyAsync(), phương thức này sẽ giúp chúng ta có thể lấy được kết quả sau khi hoàn thành tác vụ.

Tham số của phương thức supplyAsync() là một đối tượng Supplier nên chúng ta cần sửa lại đối tượng Calculator thành một đối tượng implement interface Supplier.

package com.huongdanjava.javaexample;

import java.util.function.Supplier;

public class Calculator implements Supplier<Integer> {
    
    private int a;
    private int b;

    public Calculator(int a, int b) {
        this.a = a;
        this.b = b;
    }

    @Override
    public Integer get() {
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return a + b;
    }
}

package com.huongdanjava.javaexample;

import java.util.function.Supplier;

public class Calculator implements Supplier<Integer> {

private int a;

private int b;

public Calculator(int a, int b) {

this.a = a;

this.b = b;

}

@Override

public Integer get() {

try {

Thread.sleep(3000);

} catch (InterruptedException e) {

e.printStackTrace();

}

return a + b;

}

Và lúc này trong main class chúng ta không cần phải sử dụng Executor Framework nữa bởi vì CompletableFuture sẽ thực hiện tác vụ của chúng ta trong một Thread khác rồi. Main class chúng ta lúc này có thể viết lại như sau:

package com.huongdanjava.javaexample;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class Example {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // Create new Calculator object
        Calculator c = new Calculator(2, 3);

        CompletableFuture<Integer> f = CompletableFuture.supplyAsync(c);

        System.out.println("End...");

        Thread.sleep(10000);
    }
}

package com.huongdanjava.javaexample;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.ExecutionException;

public class Example {

public static void main(String[] args) throws ExecutionException, InterruptedException {

// Create new Calculator object

Calculator c = new Calculator(2, 3);

CompletableFuture<Integer> f = CompletableFuture.supplyAsync(c);

System.out.println("End...");

Thread.sleep(10000);

}

Để có thể lấy kết quả sau khi thực hiện xong tác vụ trong phương thức get() của đối tượng Calculator và in kết quả ra console, mình sẽ sử dụng phương thức thenAccept() của CompletableFuture với tham số là một đối tượng Consumer lưu giữ giá trị sau khi tác vụ được xử lý xong.

Code của main class lúc này sẽ như sau:

package com.huongdanjava.javaexample;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class Example {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // Create new Calculator object
        Calculator c = new Calculator(2, 3);

        CompletableFuture<Integer> f = CompletableFuture.supplyAsync(c);

        f.thenAccept(result -> {
            System.out.println(result);
        });

        System.out.println("End...");

        Thread.sleep(10000);
    }
}

package com.huongdanjava.javaexample;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.ExecutionException;

public class Example {

public static void main(String[] args) throws ExecutionException, InterruptedException {

// Create new Calculator object

Calculator c = new Calculator(2, 3);

CompletableFuture<Integer> f = CompletableFuture.supplyAsync(c);

f.thenAccept(result -> {

System.out.println(result);

});

System.out.println("End...");

Thread.sleep(10000);

}

Khi chạy đoạn code trên, các bạn sẽ thấy dòng “End…” sẽ được in ra trước. Sau 3s, sau khi xử lý xong tác vụ, kết quả được trả về và gán cho biến result. Lúc này, kết quả cũng sẽ được in ra.

Rõ ràng các bạn thấy, chương trình của chúng ta không còn bị block nữa.

3/5 - (1 bình chọn)

Khanh Nguyen

Share this post

Tags

Giới thiệu về CompletableFuture trong Java

Add Comment Hủy

Khanh Nguyen

Share this post

Tags

You might also like...

Virtual thread trong Java

Scoped Value trong Java

ThreadLocal trong Java

Structured Concurrency trong Java

Add Comment Hủy