멀티 쓰레드

하나의 프로그램 안에서 여러 작업 흐름을 동시에 실행하는 방식
프로그램이 일을 처리할 때 직원 1명이 일하는 구조가 단일 쓰레드
직원 여러 명이 일을 나눠서 하는 구조가 멀티쓰레드

쓰레드란?

프로그램을 실행하면 보통 프로세스가 만들어짐
프로세스 안에서 실제로 코드를 실행하는 작업 단위가 쓰레드
프로세스는 실행 중인 프로그램 전체, 쓰레드는 그 안에서 움직이는 작업자

workerCount와 queueCapacity 를 변수로 스레드 수와 큐 사이즈 정해둠

1. BlockingQueue (작업 대기열/컨베이어 벨트)

BlockingQueue<WorkItem<T>> queue = new ArrayBlockingQueue<>(queueCapacity);

queueCapacity 크기만큼의 실제 바구니(큐)를 생성
ArrayBlockingQueue는 스레드 안전(Thread-safe)한 자료구조
동작 원리 : 메인 스레드가 이 바구니에 일감을 계속 던져 넣고(공급), 일꾼(워커 스레드)들은 바구니에서 일감을 하나씩 빼서 처리함
만약 바구니가 꽉 차면 메일 스레드는 잠시 기다리고, 바구니가 텅 비면 일꿀들도 잠시 기다리는 식의 컨베이어 벨트 역할을 함

2. AtomicInteger (처리 완료 카운터)

AtomicInteger processed = new AtomicInteger(0);

숫자 0부터 시작하는 카운터
일반적인 int는 여러 일꾼이 동시에 숫자가 누락될 수 있음
하지만 AtomicInteger는 내부적으로 동기화 처리가 되어 있어서 10명의 일꾼이 동시에 +1을 하더라도 절대 숫자가 꼬이지 않고 정확하게 누적 카운트를 해주는 안전한 카운터

3. AtomicReference (치명적 에러 공유기 / 긴급 정지 버튼)

AtomicReference<FatalStorageException> fatalError = new AtomicReference<>();
AtomicReference<RuntimeException> runtimeError = new AtomicReference<>();

여러 일꾼 중 단 한명이라도 치명적인 에러 (예 : 네트워크 드라이브 끊어짐) 를 만났을 때, 그 에러 상황을 모든 스레드에게 즉시 공유하기 위한 변수
한 일꾼이 에러를 겪고 이 변수에 에러 객체를 담아두면 메인 스레드나 다른 일꾼들이 다음 일감을 꺼내기 전에 이 변수를 슬쩍 확인해 보고 모든 작업을 즉시 중지하고 배치를 즉시 멈출 수 있게 해주는 일종의 공장 비상정지 버튼 역할을 함

위에서 준비한 설정들을 바탕으로 실제 일꾼(스레드)들을 고용하고 일을 시작시키는 부분

4. 시간 측정 시작

long batchStarted = System.nanoTime();

작업이 총 얼마나 걸렸는지 나중에 계산하기 위해 스톱워치의 시작 버튼을 누르는 역할

5. 일꾼 관리 대장 만들기

Thread[] workers = new Thread[workerCount];

고용할 일꾼들의 목록을 관리하기 위해 빈 명부를 만듦.

6. 일꾼들 고용 및 업무 지시

for (int i = 0; i < workerCount; i++) {
    final int workerNo = i + 1;
    workers[i] = new Thread(() -> consumeQueue(queue, processor, processed, fatalError, runtimeError, logPrefix),
        logPrefix + "-worker-" + workerNo);
    workers[i].start();
}

지정된 일꾼 수만큼 반복하면서 다음 작업을 수행
1. 번호 부여 : workerNo = i + 1; 을 통해 일꾼들에게 1번, 2번, 3번 번호표를 줌
2. 업무 할당 (new Thread(...)) :

• 새로운 일꾼(스레드)를 만듦
• consumeQueue라는 메서들를 따라서 바구니(queue)에서 일감을 꺼내서 처리하라고 업무 지시 내림(이때 에러 나면 비상버튼 누르라고 공유 변수들도 다 줌)

3. 명찰 달아주기 (logPrefix + "-worker-" + workerNo) :

• 스레드의 이름을 분리 보관-worker-1, 분리 보관-worker-2 처러 명확하게 붙여줌. 이렇게 하면 나중에 에러 로그를 볼 때 어떤 일꾼이 일하다가 에러 났는지 파악하기 쉬움

4. 작업 시작 (workers[i].start()) :

• 모든 준비를 마친 일꾼을 현장에 투입하고 코드가 실행되는 순간부터 각 일꾼들은 메인 스레드와 별개로 각자 알아서 컨베이어 벨트(바구니)에서 일감을 빼서 처리하기 시작함.

이 블록이 끝나면 메인 스레드 1개와 일을 열심히 하고 있느 워커 스레드 N개가 동시에 돌아가게 되는 것

consumeQueue() 메서드에서 하는 일

consumeQueue()

업무 내용

1. 무한 대기하며 일감 받기

while (true) {
    WorkItem<T> work = queue.take();

• 일꾼은 퇴근 지시가 떨어질 대까지 while(true) 계속 일함
• queue.take() : 컨베이어 벨트(바구니)에서 일감을 하나 꺼냄. 만약 바구니가 비어있다면, 새로운 일감이 들어올 때까지 그 자리에 서서 대기

2. 퇴근 지시(Poison Pill) 확인

    if (work.poison) {
        return;
    }

• Poison Pill(독약) 패턴이라고 불리는 아주 유명한 스레드 종료 기법
• 메인 스레드가 진짜 일감을 다 던져준 후, 마지막에 일꾼 수만큼 퇴근해라(poison=true)라는 가짜 일감을 던짐
• 일꾼이 바구니에서 꺼낸 일감이 만약 이 가짜 일감(퇴근 지시)이라면, return을 호출해 함수를 빠져나가고 스레드는 완전히 종료(퇴근)

3. 비상 정지(에러) 확인

    if (fatalError.get() != null || runtimeError.get() != null) {
        continue;
    }

• 본격적인 일을 시작하기 직전, 아까 공유받았던 비상 정지 버튼(에러 변수)을 슬쩍 봄
• 만약 다른 일꾼이 이미 치명적인 에러를 냈다면 지금 손에 든 일감은 처리하지 않고 바로 다음 바퀴(continue)로 넘어가서 남은 일감들을 빠르게 비우고 퇴근 지시를 기다림

4. 실제 작업 수행 및 성공 보고

    try {
        processor.accept(work.item);
        processed.incrementAndGet();

• 비상 상황이 아니라면 processor.accept(...)를 호출
  이 코드가 바로 서비스 클래스에서 넘겨준 processOne() 메서드를 실행하는 부분 (실제 파일 복사/이동, DB 업데이트 등)
• 작업이 에러 없이 무사히 끝났다면, 안전한 공유 카운터(processed)의 숫자를 +1 올려서 저 1건 무사히 끝냈습니다 라고 보고

5. 내가 일하다가 에러가 났을 때 (비상벨 누르기)

    } catch (FatalStorageException e) {
        fatalError.compareAndSet(null, e);
    } catch (RuntimeException e) {
        runtimeError.compareAndSet(null, e);
        // ...

• 만약 일하는 도중(파일 복사 중)에 랜선이 뽑히거나 스토리지 서버가 죽어서 FatalStorageException이 발생했다면?
• compareAndSet(null, e)를 호출해서 비상 정지 버튼을 꽉 누름. (아무도 에러를 안 냈을 때만 내 에러를 등록함)
• 이렇게 버튼을 누르면 옆에서 일하던 다른 일꾼들도 3번 과정에서 에러를 감지하고 즉시 작업을 멈추게 됨

일꾼이 바구니에 일감을 꺼내서 에러가 없는지 보고 배정받은 실제 파일 처리(processOne)를 하고 성공하면 카운트를 올리고 치명적 에러가 나면 전체 스탑 버튼을 누르는 사이클