Thực thi nhiều tác vụ cùng lúc như thế nào trong Java?

Trong bài viết về CompletableFuture, chúng ta đã tìm hiểu về cách sử dụng multi-thread trong Java 8. Trong bài này, chúng ta sẽ cùng xem cách sử dụng CompletableFuture trong một bài toán thực tế.

Giả sử chúng ta có một ứng dụng cần thực hiện 2 công việc, tạm gọi là work1 và work2. Có hàng nghìn hàng triệu công việc work1 và khi mỗi công việc work1 hoàn thành sẽ có một danh sách các công việc work2 cần thực hiện. Để tiết kiệm được thời gian, chúng ta sẽ sử dụng Multi-Thread để thực thi công việc work1, khi mỗi công việc work1 hoàn thành, chúng ta cũng sẽ sử dụng Multi-Thread để xử lý kết quả nhận được từ công việc work1 (thực thi công việc work2).

Chương trình của chúng ta như sau:

package com.maixuanviet.completable_future;
 
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
 
/**
 * All level is running with multi-thread
 */
public class ConcurrencyWithCompletableFuture3 {
 
    public static void main(String[] args) {
        List<String> works = new ArrayList<>();
        works.add("A");
        works.add("B");
        works.add("C");
        works.add("D");
        works.add("E");
        runMultipleAsync(works);
    }
 
    private static void runMultipleAsync(List<String> works) {
        List<CompletableFuture<List<Void>>> allOfWork1Futures = new ArrayList<>();
        works.stream().forEach(work -> {
            allOfWork1Futures.add(createWork1(work).thenCompose(work1Results -> {
                List<CompletionStage<Void>> allOfWork2Futures = work1Results.stream()
                        .map(work1Result -> createWork2(work1Result)).collect(Collectors.toList());
                CompletableFuture<Void> done = CompletableFuture
                        .allOf(allOfWork2Futures.toArray(new CompletableFuture[allOfWork2Futures.size()]));
                return done.thenApplyAsync(v -> allOfWork2Futures.stream().map(CompletionStage::toCompletableFuture)
                        .map(CompletableFuture::join) // Returns the result value when complete
                        .collect(Collectors.toList()));
            }).whenCompleteAsync((result, th) -> {
                // Do something when complete
            }).toCompletableFuture());
        });
        CompletableFuture<Void> done = CompletableFuture
                .allOf(allOfWork1Futures.toArray(new CompletableFuture[allOfWork1Futures.size()]))
                .whenComplete((result, th) -> {
                    // Do something when complete
                });
        done.join(); // Returns the result value when complete
    }
 
    private static CompletionStage<List<String>> createWork1(String str) {
        return CompletableFuture.completedFuture(str).thenApplyAsync(s -> executeWork1(s));
    }
 
    private static CompletionStage<Void> createWork2(String str) {
        return CompletableFuture.completedFuture(str).thenAcceptAsync(s -> executeWork2(s));
    }
 
    private static List<String> executeWork1(String _item) {
        waitingForComplete();
        System.out.println("Work" + _item + " -> work1");
        return Arrays.asList(_item + "_item" + 1, _item + "_item" + 2);
    }
 
    private static void executeWork2(String data) {
        waitingForComplete();
        System.out.println("Work" + data + " -> work2");
    }
 
    private static void waitingForComplete() {
        try {
            TimeUnit.SECONDS.sleep(random(0, 3));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
 
    private static int random(int min, int max) {
        Random r = new Random();
        return r.nextInt((max - min) + 1) + min;
    }
}

Output của chương trình:

WorkB -> work1
WorkC -> work1
WorkB_item1 -> work2
WorkA -> work1
WorkB_item2 -> work2
WorkC_item1 -> work2
WorkD -> work1
WorkE -> work1
WorkA_item1 -> work2
WorkC_item2 -> work2
WorkD_item1 -> work2
WorkD_item2 -> work2
WorkE_item2 -> work2
WorkA_item2 -> work2
WorkE_item1 -> work2
5.0