대용량 데이터 처리 방식 과 흐름

1. 코드 실행 순서 및 구조

대용량 데이터를 처리하는 방식은 일반적으로 다음과 같은 단계를 거칩니다.

1. 데이터 로드 (Load Data)
   • 데이터베이스, 파일, API, 메시지 큐(Kafka 등)에서 데이터를 읽어옵니다.
2. 데이터 분할 (Partitioning)
   • 전체 데이터를 한 번에 처리하면 성능이 저하되므로, 여러 개의 작은 청크(batch)로 나눕니다.
3. 병렬 처리 (Parallel Processing)
   • 각 청크를 개별 스레드에서 병렬로 처리합니다.
   • 스레드 풀(Thread Pool)을 활용하여 일정 개수의 스레드만 실행되도록 제어합니다.
4. 비동기 실행 (Asynchronous Execution)
   • 비동기 방식으로 데이터를 처리하여 블로킹(blocking)을 최소화합니다.
5. 데이터 저장 (Save Data)
   • 처리된 데이터를 데이터베이스 또는 외부 시스템에 저장합니다.

---

 

2. 코드 예제 및 실행 흐름

 

(1) 스레드 풀을 활용한 대량 데이터 처리

 

다음 코드는 10개의 스레드로 데이터를 병렬 처리하는 예제입니다.

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class LargeDataProcessor {

    private static final int THREAD_POOL_SIZE = 10;

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

        List<String> dataList = fetchData();  // 대량 데이터 가져오기
        int batchSize = dataList.size() / THREAD_POOL_SIZE;

        for (int i = 0; i < THREAD_POOL_SIZE; i++) {
            int start = i * batchSize;
            int end = (i == THREAD_POOL_SIZE - 1) ? dataList.size() : (i + 1) * batchSize;
            List<String> batch = dataList.subList(start, end);

            executorService.submit(() -> processBatch(batch));
        }

        executorService.shutdown(); // 모든 작업이 끝나면 스레드 풀 종료
    }

    private static List<String> fetchData() {
        // 데이터베이스 또는 파일에서 데이터 가져오는 로직
        return List.of("data1", "data2", "data3", "data4", "data5"); // 예제 데이터
    }

    private static void processBatch(List<String> batch) {
        for (String data : batch) {
            System.out.println(Thread.currentThread().getName() + " processing: " + data);
        }
    }
}

실행 순서

1. fetchData()가 데이터를 가져옵니다.
2. 데이터를 THREAD_POOL_SIZE만큼 분할합니다.
3. 각 청크(batch)를 스레드 풀을 이용해 병렬로 처리합니다.
4. 모든 데이터 처리가 끝나면 스레드 풀을 종료합니다.

---

 

(2) Spring Boot 기반 비동기 데이터 처리

 

비동기 방식으로 데이터를 처리하면, 스레드가 하나의 요청에 블로킹되지 않고 동시에 여러 요청을 처리할 수 있습니다.

 

비동기 서비스 구현

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.util.concurrent.CompletableFuture;

@Service
public class AsyncDataProcessingService {

    @Async
    public CompletableFuture<String> processData(String data) {
        System.out.println(Thread.currentThread().getName() + " processing: " + data);
        return CompletableFuture.completedFuture("Processed: " + data);
    }
}

 

비동기 호출

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.CompletableFuture;

@RestController
public class DataProcessingController {

    @Autowired
    private AsyncDataProcessingService asyncService;

    @GetMapping("/process")
    public CompletableFuture<String> process() {
        return asyncService.processData("Example Data");
    }
}

 

실행 순서

1. 클라이언트가 /process 엔드포인트를 호출합니다.
2. processData()가 @Async에 의해 별도의 스레드에서 실행됩니다.
3. 요청을 보낸 클라이언트는 즉시 응답을 받으며, 백그라운드에서 데이터 처리가 진행됩니다.

---

 

3. 성능 최적화를 위한 추가 기법

 

(1) Kafka를 활용한 대량 데이터 스트리밍

Kafka는 대량의 데이터를 비동기적으로 처리할 수 있는 메시지 큐 시스템입니다.

 

Kafka Listener 설정

@KafkaListener(topics = "data_topic", groupId = "data_group")
public void listen(String message) {
    processMessage(message);
}

Kafka를 사용하면 메시지를 비동기적으로 받아서 처리할 수 있습니다.

 

(2) 데이터 분할 및 병렬 처리

100,000개의 데이터를 한 번에 처리하는 대신, 1,000개씩 나누어 병렬로 처리하면 성능이 향상됩니다.

int batchSize = 1000;
for (int i = 0; i < totalDataSize; i += batchSize) {
    List<Data> batch = fetchData(i, batchSize);
    executorService.submit(() -> processBatch(batch));
}

---

 

4. 결론

 

대용량 데이터를 효과적으로 처리하려면 다음과 같은 기법을 활용해야 합니다.

• 스레드 풀 활용: Executors.newFixedThreadPool(n)을 사용하여 병렬 처리
• 비동기 처리: @Async를 활용하여 블로킹 최소화
• 데이터 분할 처리: 작은 청크로 나누어 처리
• Kafka 등 메시지 큐 활용: 스트리밍 데이터 처리를 위한 비동기 큐 시스템

이러한 기법을 조합하면 시스템 성능을 최적화하고 안정성을 높일 수 있습니다.