비동기와 동기(Thread, synchronized, CompletableFuture)

By pink_salt

프로그래밍 언어/Java

2024. 11. 22. 03:13

728x90

동기란? - Synchronous

동기는 작업이 순차적으로 실행되는 방식을 의미한다. 현재 작업이 끝나야 다음 작업을 실행할 수 있다.

작업의 순차 실행을 보장한다.
호출자가 결과를 즉시 확인할 수 있다.

비동기란? - Asynchronous

비동기는 작업이 독립적으로 실행되는 방식을 의미한다. 작업 요청 후 결과를 기다리지 않고, 다른 작업을 동시에 처리할 수 있다.

비동기와 동기 비교

특징	동기(Synchronous)	비동기(Asynchronous)
작업 처리 방식	순차적으로 작업 처리	독립적으로 작업 처리
작업 대기 여부	호출자가 작업 완료까지 대기	호출자가 대기하지 않고 다른 작업 수행
사용 사례	트랜잭션 처리, 데이터베이스 작업	네트워크 요청, 파일 I/O 처리
장점	간단하고 구현이 쉬움	작업 병렬 처리로 성능 향상
단점	작업 대기 시간이 길어질 수 있음	구현이 복잡하고 디버깅이 어려움

자바에서는 비동기와 동기를 멀티스레드와 병렬 프로그래밍에서 자주 사용한다.

동기 방식: Thread, synchronized를 사용해 작업의 순차성과 스레드 안전성을 보장.
비동기 방식: CompletableFuture, ExecutorService 등을 사용해 작업의 비동기 처리를 구현.

트랜잭션 처리: 동기의 안정성

트랜잭션은 작업의 순차 실행과 데이터의 무결성이 중요하므로 동기 방식이 적합하다.
예시 :
- 데이터베이스 업데이트: 동기적으로 실행하여 작업의 순서를 보장.
- 파일 작성: 파일 쓰기 작업이 완료된 후 다음 작업을 실행.

public class SyncTransactionExample {
    public static void main(String[] args) {
        performTransaction();
        System.out.println("트랜잭션 완료 후 다른 작업 수행");
    }

    private static void performTransaction() {
        synchronized (SyncTransactionExample.class) {
            System.out.println("트랜잭션 처리 중...");
            // 트랜잭션 작업
            try {
                Thread.sleep(2000); // 작업 대기 시뮬레이션
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
/*
트랜잭션 처리 중...
트랜잭션 완료 후 다른 작업 수행
*/

위 코드는 Java의 synchronized 키워드를 사용해 동기화된 방식으로 트랜잭션을 처리하는 예제이다.

main 메서드 실행:
- performTransaction() 메서드를 호출하여 트랜잭션 작업을 시작.
동기화 블록 진입:
- 현재 스레드가 SyncTransactionExample.class의 락을 획득.
- 동기화 블록 내부에서 트랜잭션 작업 실행.
트랜잭션 작업:
- "트랜잭션 처리 중..." 출력.
- 2초 동안 대기(Thread.sleep(2000)).
동기화 블록 종료:
- 작업 완료 후 락 해제.
- 다른 스레드가 동기화 블록에 진입 가능.
다음 작업 실행:
- "트랜잭션 완료 후 다른 작업 수행" 출력.

네트워크 요청: 비동기의 강점

네트워크 요청은 대기 시간이 길어질 가능성이 크므로, 작업이 완료될 때까지 대기하지 않는 비동기 방식이 적합하다.
예시 :
- REST API 호출: Spring에서 비동기적으로 외부 API를 호출하여 응답을 처리.
- 데이터 수집: 비동기로 여러 데이터 소스를 병렬로 호출하여 처리 시간을 단축.

import java.util.concurrent.CompletableFuture;

public class AsyncNetworkExample {
    public static void main(String[] args) {
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            // 네트워크 요청 시뮬레이션
            simulateNetworkRequest();
            return "응답 완료";
        }).thenAccept(response -> System.out.println("비동기 응답: " + response));

        System.out.println("다른 작업 수행 중...");

        future.join(); // 비동기 작업이 끝날 때까지 대기
    }

    private static void simulateNetworkRequest() {
        try {
            Thread.sleep(2000); // 네트워크 요청 대기 시간
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
/*
다른 작업 수행 중...
비동기 응답: 응답 완료
*/

위 코드는 Java의 CompletableFuture를 사용해 비동기적으로 네트워크 요청을 처리하는 방식이다.

메인 스레드
- CompletableFuture.supplyAsync()를 호출하여 비동기 작업을 시작.
- "다른 작업 수행 중..."을 출력.
ForkJoinPool의 작업 스레드
- supplyAsync에 전달된 람다식을 실행.
- simulateNetworkRequest()를 호출하여 2초 동안 대기.
- "응답 완료" 문자열을 반환.
결과 처리
- 작업 완료 후 thenAccept에 정의된 코드가 실행되어 결과를 콘솔에 출력.

CompletableFuture에 대해서는 아래에서 더 자세히 알아보겠다.

자바에서 비동기와 동기 처리를 해보자.

1. Thread와 synchronized (동기 처리)

Thread란?

프로그램 내에서 병렬 작업을 처리하기 위한 실행 단위.
Java에서 기본적으로 Main Thread가 존재하며, 여러 스레드를 생성하여 동시에 작업을 실행할 수 있다.

공유 자원 문제?

여러 스레드가 동시에 동일한 자원(변수, 데이터 구조 등)에 접근하면, 데이터의 무결성이 깨지는 문제가 발생할 수 있다.
예: 두 스레드가 동시에 하나의 변수 값을 증가시키면, 원하는 값 대신 잘못된 결과가 나올 수 있음.

synchronized란?

synchronized는 스레드 간 동기화를 보장하는 키워드.
특정 블록 또는 메서드가 한 번에 하나의 스레드만 접근 가능하도록 제한한다.
이를 통해 공유 자원의 데이터 무결성을 유지할 수 있다.

동기화가 필요한 상황을 보겠다.

package test;

public class ThreadWithoutSync {
    private static int counter = 0;

    public static void increment() {
        counter++;
    }

    public static void main(String[] args) {
        Runnable task = () -> {
            for (int i = 0; i < 1000; i++) {
                increment(); 
            }
        };

        Thread t1 = new Thread(task);
        Thread t2 = new Thread(task);

        t1.start();
        t2.start();

        try {
            t1.join();
            t2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("최종 counter 값: " + counter);
    }
}
// 최종 counter 값: 1771 또는 다른 값!
// 계속 2000이 일관되게 나오지 않는다.

두 스레드가 동시에 counter를 읽고 업데이트하면 값이 덮어씌워지는 문제가 발생하여 최종 값이 2000이 출력되지 않는다.

package test;

public class ThreadWithSync {
    private static int counter = 0;

    // synchronized 키워드로 메서드 동기화
    public static synchronized void increment() {
        counter++;
    }

    public static void main(String[] args) {
        Runnable task = () -> {
            for (int i = 0; i < 1000; i++) {
                increment(); // 동기화된 메서드 호출
            }
        };

        Thread t1 = new Thread(task);
        Thread t2 = new Thread(task);

        t1.start();
        t2.start();

        try {
            t1.join();
            t2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 항상 2000이 출력됨
        System.out.println("최종 counter 값: " + counter);
    }
}
// 최종 counter 값: 2000

increment() 메서드에 synchronized를 추가하여 동시에 한 스레드만 해당 메서드에 접근 가능하도록 제한하니 2000으로 값이 잘 나오는 것을 확인할 수 있다.

synchronized 사용 방식

블록 또는 메서드를 동기화하여, 한 번에 하나의 스레드만 해당 메서드에 접근 가능하도록 한다.

public synchronized void method() {
    // 메서드 전체가 동기화됨
}

public void method() {
    synchronized (this) { // 또는 특정 객체
        // 동기화된 블록
    }
}

메서드 전체를 동기화하여, 한 번에 하나의 스레드만 해당 메서드에 접근 가능.
특정 코드 블록만 동기화하여, 불필요한 성능 저하를 방지.
락(Lock) 객체를 사용하여 동기화를 구현.

예시) 은행 계좌 잔액 관리 -> 여러 스레드가 하나의 계좌에서 입출금을 동시에 처리하면 잔액이 잘못 계산될 수 있다.

이를 위해 synchronized를 활용한 코드이다.

public class BankAccount {
    private int balance;

    public BankAccount(int initialBalance) {
        this.balance = initialBalance;
    }

    // 동기화된 입금 메서드
    public synchronized void deposit(int amount) {
        balance += amount;
        System.out.println(Thread.currentThread().getName() + " 입금: " + amount + ", 잔액: " + balance);
    }

    // 동기화된 출금 메서드
    public synchronized void withdraw(int amount) {
        if (balance >= amount) {
            balance -= amount;
            System.out.println(Thread.currentThread().getName() + " 출금: " + amount + ", 잔액: " + balance);
        } else {
            System.out.println(Thread.currentThread().getName() + " 출금 실패: 잔액 부족");
        }
    }

    public static void main(String[] args) {
        BankAccount account = new BankAccount(1000);

        Runnable task1 = () -> {
            for (int i = 0; i < 3; i++) account.deposit(100);
        };

        Runnable task2 = () -> {
            for (int i = 0; i < 3; i++) account.withdraw(100);
        };

        Thread t1 = new Thread(task1, "Thread-1");
        Thread t2 = new Thread(task2, "Thread-2");

        t1.start();
        t2.start();
    }
}
/*
Thread-1 입금: 100, 잔액: 1100
Thread-1 입금: 100, 잔액: 1200
Thread-1 입금: 100, 잔액: 1300
Thread-2 출금: 100, 잔액: 1200
Thread-2 출금: 100, 잔액: 1100
Thread-2 출금: 100, 잔액: 1000

*/

synchronized의 주의점

성능 저하:
- 동기화는 추가적인 비용(락 획득 및 해제)을 발생시키므로, 필요하지 않은 경우 피해야 함.
데드락(Deadlock):
- 여러 스레드가 서로 다른 락을 기다릴 때 교착 상태가 발생할 수 있음.

그렇기 때문에 공유 자원 보호가 필요한 경우만 synchronized를 사용해야 한다. 또한 동기화 문제를 잘 이해하여 레이스 컨데션, 데드락 문제가 발생하지 않도록 해야 한다.

참고로, 단순 동기화 작업은 Atomic 클래스나 ReentrantLock 과 같은 대체 방법이 있다고 한다. 해당 내용은 추후에 더 살펴보겠다.

2. CompletableFuture란? (비동기 처리)

CompletableFuture는 Java 8에서 도입된 클래스로, 비동기 작업을 처리하고 결과를 다룰 수 있는 기능을 제공한다.

기존 Future의 문제

여러 연산을 결합하기 어려운 문제
비동기 처리 중에 발생하는 예외를 처리하기 어려운 문제

해당 문제점을 개선하기 위해 CompletableFuture 클래스는 java5에 추가된 Future 인터페이스와 CompleteStage 인터페이스를 구현하고 있다.

Future: java5에서 비동기 연산을 위해 추가된 인터페이스
CompleteStage: 여러 연산을 결합할 수 있도록 연산이 완료되면 다음 단계의 작업을 수행하거나 값을 연산하는 비동기식 연산 단계를 제공하는 인터페이스

Future	CompletableFuture
Blocking	Non-blocking
여러 연산을 함께 연결하기 어려움	여러 연산을 함께 연결
여러 연산 결과를 결합하기 어려움	여러 연산 결과를 결합
연산 성공 여부만 확인할 수 있고 예외 처리의 어려움	exceptionally(), handle()을 통한 예외 처리

CompletableFuture의 장점?

비동기 작업 처리: 별도의 스레드에서 작업을 실행하고, 결과가 준비되면 콜백을 통해 처리.
결과 처리의 간소화: 작업 완료 후 후속 작업을 자연스럽게 연결 가능 (메서드 체인).
병렬 작업 처리: 여러 비동기 작업을 병렬로 실행하고, 결과를 합칠 수 있음.
예외 처리: 비동기 작업 중 발생한 예외를 간단히 처리 가능.

단점?
- 복잡성 증가, 디버깅의 어려움, 가독성 감소

아래 예시들은 CompletableFuture의 사용 방법을 간단하게 표현한 것들이다.

일단, CompletableFuture 클래스의 정적 메서드인 supplyAsync() 메서드를 통해 CompletableFuture 인스턴스를 생성할 수 있다.

ForkJoinPool?
자바의 병렬 처리를 위한 스레드 풀이다.
작업을 여러 개의 작은 서브 작업(Subtask)으로 분할(Fork)하고, 각 서브 작업을 병렬로 실행한다.
서브 작업의 결과를 합쳐(Join) 최종 결과를 반환한다.

import java.util.concurrent.CompletableFuture;

public class CompletableFutureExample {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 비동기 작업
            return "Hello, CompletableFuture!";
        });

        // 작업 완료 후 결과 처리
        future.thenAccept(result -> System.out.println("결과: " + result));
    }
}
// 결과: Hello, CompletableFuture!

supplyAsync(Supplier<T>)
- Supplier를 인수로 supplyAsync()를 호출하면 ForkJoinPool.commonPool()에서 전달된 Supplier를 비동기적으로 호출한 뒤 CompleteableFuture 인스턴스를 반환하게 된다.
thenAccept(Consumer<T>)
- 작업 완료 후 결과를 소비.
- Consumer를 인자로 받고, 결과를 CompletableFuture<Void>를 반환한다.
- 리턴 타입이 없는 로직을 호출할 때 사용할 수 있다.

연산 결합 과정을 나타낸 예시이다.

import java.util.concurrent.CompletableFuture;

public class CombineExample {
    public static void main(String[] args) {
        CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> "작업 1 결과");
        CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> "작업 2 결과");

        // 두 작업 결과를 조합
        task1.thenCombine(task2, (result1, result2) -> result1 + " + " + result2)
             .thenAccept(System.out::println);
    }
}
// 작업 1 결과 + 작업 2 결과

thenCombine()

메서드로는 두 개의 독립적인 Future를 처리하고 두 결과를 결합하여 추가적인 작업을 수행할 수 있다.

thenCompose()

메서드는 두 개의 Future를 순차적으로 연결한다.

예외 처리를 한 예시이다.

import java.util.concurrent.CompletableFuture;

public class ExceptionHandlingExample {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> {
            if (Math.random() > 0.5) throw new RuntimeException("작업 실패!");
            return "작업 성공";
        }).exceptionally(ex -> {
            System.out.println("예외 처리: " + ex.getMessage());
            return "기본 값";
        }).thenAccept(result -> System.out.println("결과: " + result));
    }
}

exceptionally()

작업 중 발생한 예외를 처리하고 기본 값을 반환.

체이닝 예시이다.

import java.util.concurrent.CompletableFuture;

public class AsyncChainingExample {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> "작업 시작")
                         .thenApply(result -> result + " → 중간 처리")
                         .thenApply(result -> result + " → 최종 처리")
                         .thenAccept(System.out::println);
    }
}
// 작업 시작 → 중간 처리 → 최종 처리

thenApply(Function<T, R>)

결과를 변환하여 다음 단계로 전달.
Function의 반환 값을 가지고 있는 CompletableFuture<U>을 반환
이전 단계의 결괏값을 인수로 사용

각 CompletableFuture의 호출 결과가 필요할 경우, thenApply() 메서드를 사용하는 것이 적합
각 CompletableFuture의 결과를 결합한 최종 연산 결과만 필요한 경우, thenCompose() 메서드를 사용하는 것이 적합

병렬 처리 예시이다.

allOf() 정적 메서드를 사용하면 여러 Future를 병렬로 처리할 수 있다. Future의 처리를 대기하다가 모두 완료 되면 CompletableFuture<Void>를 반환

package test;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class AllOfExample {
    public static void main(String[] args) {
        // 병렬로 실행할 작업 정의
        CompletableFuture<String> api1 = CompletableFuture.supplyAsync(() -> simulateApiCall("API1", 2));
        CompletableFuture<String> api2 = CompletableFuture.supplyAsync(() -> simulateApiCall("API2", 3));
        CompletableFuture<String> api3 = CompletableFuture.supplyAsync(() -> simulateApiCall("API3", 1));

        // allOf로 병렬 처리
        CompletableFuture<Void> allOf = CompletableFuture.allOf(api1, api2, api3);

        // 모든 작업 완료 후 결과 처리
        allOf.thenRun(() -> {
            try {
                System.out.println("모든 API 호출 완료");
                System.out.println("API1 결과: " + api1.get());
                System.out.println("API2 결과: " + api2.get());
                System.out.println("API3 결과: " + api3.get());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).join(); // 비동기 작업이 끝날 때까지 대기
        

        System.out.println("다른 작업 수행 중...");
    }

    // API 호출 시뮬레이션
    private static String simulateApiCall(String apiName, int seconds) {
        try {
            System.out.println(apiName + " 호출 중...");
            TimeUnit.SECONDS.sleep(seconds); // 대기 시뮬레이션
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return apiName + " 응답 완료";
    }
}
/*
API1 호출 중...
API3 호출 중...
API2 호출 중...
모든 API 호출 완료
API1 결과: API1 응답 완료
API2 결과: API2 응답 완료
API3 결과: API3 응답 완료
다른 작업 수행 중...
*/

한계 : 병렬 처리는 가능하지만, 모든 Future의 결과를 결합한 결괏값을 반환할 수 없는 한계가 있다.

join() 메서드를 활용하면 allOf() 메서드의 한계를 극복할 수 있지만, Future가 정상적으로 완료되지 않을 경우 확인되지 않은 예외가 발생할 수 있는 단점이 있다는 점을 고려해야 한다.

메서드	설명	반환값
supplyAsync(Supplier)	비동기 작업 실행 및 결과 반환	비동기 작업의 결과(T)를 포함한 CompletableFuture 객체.
thenApply(Function)	이전 단계의 결과(T)를 받아 새로운 값을 계산(R)하여 반환.	변환된 결과를 포함한 CompletableFuture<R>
thenAccept(Consumer)	이전 단계의 결과(T)를 받아 처리(출력, 저장 등)하지만, 값을 반환하지 않음	CompletableFuture<Void>
thenCombine()	두 비동기 작업의 결과(T, U)를 조합하여 새로운 값을 생성(R).	조합된 결과를 포함한 CompletableFuture<R>
exceptionally()	작업 중 예외가 발생했을 때 이를 처리하고 기본 값을 반환	예외를 처리하거나 복구된 값을 포함한 CompletableFuture<T>
allOf()	여러 작업을 병렬로 실행하고, 모두 완료될 때 실행	CompletableFuture<Void>

REST API 병렬 호출 예시 : 여러 외부 API를 동시에 호출하고 결과를 조합한 경우

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.concurrent.CompletableFuture;

public class RestExample {
    public static void main(String[] args) {
        // API 요청을 CompletableFuture로 실행
        CompletableFuture<String> api1 = CompletableFuture.supplyAsync(() -> callApi("https://api.frankfurter.app/latest"));
        CompletableFuture<String> api2 = CompletableFuture.supplyAsync(() -> callApi("https://api.frankfurter.app/2024-01-01"));

        // 모든 작업이 완료될 때 실행
        CompletableFuture.allOf(api1, api2).thenRun(() -> {
            try {
                System.out.println("API1 결과: " + api1.get());
                System.out.println("API2 결과: " + api2.get());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).join(); // 비동기 작업 완료 대기

        System.out.println("모든 작업이 완료되었습니다.");
    }

    // HTTP 요청을 수행하는 메서드
    private static String callApi(String url) {
        StringBuilder result = new StringBuilder();
        try {
            // URL 연결 설정
            URL apiUrl = new URL(url);
            HttpURLConnection connection = (HttpURLConnection) apiUrl.openConnection();
            connection.setRequestMethod("GET");

            // 응답 코드 확인
            int responseCode = connection.getResponseCode();
            if (responseCode == HttpURLConnection.HTTP_OK) {
                // 응답 데이터를 읽음
                try (BufferedReader reader = new BufferedReader(new InputStreamReader(connection.getInputStream()))) {
                    String line;
                    while ((line = reader.readLine()) != null) {
                        result.append(line);
                    }
                }
            } else {
                result.append("API 호출 실패 (응답 코드: ").append(responseCode).append(")");
            }
        } catch (Exception e) {
            result.append("API 호출 중 예외 발생: ").append(e.getMessage());
        }
        return result.toString();
    }
}

참고 : rest api는 https://frankfurter.dev/ 해당 사이트에서 제공하는 것을 사용하였다.

데이터 처리 파이프라인 예시 : 대량 데이터를 읽고 처리한 후 저장하는 파이프라인 작업

import java.util.concurrent.CompletableFuture;

public class DataExample {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> readData())
                .thenApply(data -> processData(data))
                .thenAccept(result -> saveData(result));


    }
    private static String readData() { return "→ 데이터 읽기 완료 "; }
    private static String processData(String data) { return data + "→ 처리 완료"; }
    private static void saveData(String result) { System.out.println("저장 " + result); }
}
// 저장 → 데이터 읽기 완료 → 처리 완료

비동기와 동기 관련 면접 질문

동기와 비동기의 차이에 대해서 설명해주세요.

동기는 말 그대로 동시에 일어난다는 뜻입니다. 요청을 하면 시간이 얼마나 걸리던지 요청한 자리에 결과로 나와야합니다. 비동기식은 동시에 일어나지 않으며 요청과 동시에 결과가 일어나지 않을 거라는 약속입니다. 비동기식은 동기식보다 설계가 복잡하지만 결과가 주어지는 동안 다른 작업을 할 수 있다는 장점이 있습니다.

자바에서 비동기를 구현하는 방법은 무엇인가요?

CompletableFuture -> Java 8부터 제공되는 비동기 API로, 비동기 작업을 실행하고 결과를 처리하기 위한 메서드를 제공합니다.
ExecutorService -> 스레드 풀을 사용하여 비동기 작업을 관리합니다.
스레드 직접 생성 -> 기본적으로 Thread 클래스를 사용해 비동기 작업을 실행할 수도 있습니다.

비동기를 사용해야 하는 상황은 언제인가요?

비동기는 다수의 요청을 처리하거나 응답 속도가 중요한 작업에 적합합니다.
그렇기 때문에 외부 API 호출, 데이터 베이스 쿼리 등 시간이 오래 걸리는 작업에 비동기를 사용하여 응답 대기 없이 다른 작업을 수행할 수 있습니다.
또한 병렬 처리를 통해 대규모 데이터를 효율적으로 처리하는데에 비동기를 사용하면 좋습니다.

참고 :