در جاوا 8+ چرا وقتی ExecutorService هست، کلاس CompletableFuture هم وجود دارد؟

یکی از سوالات چالشی همیشه در مصاحبه ها، سوالاتی مربوط به Concurrency بوده. خصوصا مربوط به ویژگی های جاوا 8 به بالا. گاها ممکنه سوال بپرسن که ExecutorService یا CompletableFuture؟ چرا؟

مثال زیر رو در نظر بگیرید که دو عملیات دارید یکی برای اضافه کردن دو عدد و دیگری برای ضرب در 15:

public static Integer add(int a, int b) {
	return a + b;
}
public static Integer multiply(int result) {
    return result * 15;
}

در نظر بگیرید این دو عملیات رو تمایل دارید پشت سر هم انجام بدید.

در مرحله اول بریم دو عملیات جمع و ضرب رو با استفاده از ExecutorService و Future ببینیم:

public static void main(String[] args) throws ExecutionException, InterruptedException {	 
	ExecutorService executor = Executors.newFixedThreadPool(3); //1
	List<String> finalResultList = new ArrayList<>();
	for (int i = 0; i < 10; i++) {
		Future<Integer> futureResult = executor.submit(new Add(10, 20)); // 2
		Integer intermediateResult = futureResult.get(); // blocking // 3
		Future<Integer> finalResult = executor.submit(new Multiply(intermediateResult)); // 4
		finalResultList.add(i + &quot &quot + finalResult.get()); // blocking //5
	}
	executor.shutdown();
}

static class Add implements Callable<Integer> {
	int a;
	int b;

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

	@Override
	public Integer call() throws Exception {
		return a + b;
	}
}

static class Multiply implements Callable<Integer> {
	int result;
	Multiply(int result) {
		this.result = result;
	}

	@Override
	public Integer call() throws Exception {
		result = result * 10;
		System.out.println(Thread.currentThread().getName() + &quot &quot + result);
		return result;
	}
}

بریم سراغ بررسی کد:

در خط 1، یک شی از ExcutorService با تعداد Thread حداکثر 3 ایجاد کردیم. در خط 3 برای اینکه بتونیم نتایج عملیات ضرب رو داشته باشیم مجبوریم پاسخ عملیات رو در thread main داشته باشیم که این کار رو با متد get() انجام میدیم یا اصطلاحا blocking انجام میدیم و در خط 4 نتیجه رو در thread جدیدی برای عملیات ضرب ارسال میکنیم.

خروجی به صورت زیر شد:

> Task :ExecutorServiceSample.main()
pool-1-thread-2 300
pool-1-thread-1 300
pool-1-thread-3 300
pool-1-thread-2 300
pool-1-thread-1 300
pool-1-thread-3 300
pool-1-thread-2 300
pool-1-thread-1 300
pool-1-thread-3 300
pool-1-thread-2 300

همینطور که میبینید عملیات به این صورت پیش رفته که هر thread بعد از اتمام thread بعدی کارش رو شروع و thread ها به صورت همزمان نتونستن کارشون رو شروع کنن. چون ترتیب استفاده از threadها حفط شده. در نتیجه عملیات هایی که صف ExecutorService هستند به صورت Synchronize و Blocking Queue عمل میکنند و نمیتوانند همزمان عملیات ها را انجام دهند.

همانطور که در تصویر بالا میبینید ExecutorService ابتدا کار جدیدی را از صف برمیدارد و به اتمام میرساند سپس به سراغ کار بعدی می رود.

در مرحله بعدی میخواهیم همین کار را با استفاده از CompletableFuture انجام دهیم:

public static void main(String[] args) throws ExecutionException, InterruptedException {
	ExecutorService executor = Executors.newFixedThreadPool(3);
	List<CompletableFuture<Integer>> finalResultList = new ArrayList<>();

	for (int i = 0; i < 10; i++)  {
        	CompletableFuture<Integer> voidCompletableFuture = CompletableFuture 
                .supplyAsync(() -> add(10, 20), executor) //1
                .thenApplyAsync(CompletableFutureSample::multiply, executor); //2

		finalResultList.add(voidCompletableFuture);
	}

	for (CompletableFuture<Integer> future : finalResultList) {
        	future.get(); //3
	}
	executor.shutdown();
}

public static Integer add(int a, int b) {
	return a + b;
}

public static Integer multiply(int a) {
	a = a * 10;
	System.out.println(Thread.currentThread().getName() + &quot &quot + a);
	return a;
}

بریم و کد رو بررسی کنیم:

در خط 1 با استفاده از supplyAsync عملیات جمع به صورت async معرفی شده و با توجه به انکه ترتیب عملیات جمع و ضرب برای ما مهم هست، بعد از دریافت نتیجه، جواب رو بدون اینکه به Thread main منتقل کنیم، با استفاده از متد thenApplyAsync مستقیم به thread جدید که از ExecutorService درخواست کردیم ارسال میکنیم. نکته مهم این هستش که میتونیم به صورت همزمان چندین task رو شروع کنیم. حال ممکنه هر Task شامل چند عملیات باشه مثل ضرب و جمع.

خروجی رو بینیم:

pool-1-thread-1 300
pool-1-thread-1 300
pool-1-thread-3 300
pool-1-thread-1 300
pool-1-thread-3 300
pool-1-thread-1 300
pool-1-thread-3 300
pool-1-thread-1 300
pool-1-thread-3 300
pool-1-thread-2 300

همانطور که میبینید ترتیب Thread ها دیگه رعایت نشده و هر Threadی که کارش تموم شده Task بعدی رو شروع کرده به انجام.

تصویر زیر ببینید:

همانطور که در تصویر بالا میبینید، Task ها به صورت همزمان انجام شده اند نه به صورت Blocking

برای محتواهای بیشتر من رو در linkedIn و github دنبال کنید!

شاد باشید! :-)