Multithreading in Java

Executor Class

task class는 Runnable 인터페이스를 구현하여 만들 수 있다는 것을 배웠다. 이렇게 만들어진 task를 실행시키는 방법은 아래와 같다.

Runnable task = new TaskClass(task);
new Thread(task).start();

이와 같은 방식은 개발자 입장에서 다소 번거롭다. 개발자가 스레드를 직접 생성하고 관리하는 부담을 줄이기 위해 Executor 프레임워크가 등장했다.

   +---------------------------------+
   |	     <<interface>>           |
   |  java.util.concurrent.Executor  |
   +---------------------------------+
   | +execute(Runnable object): void |
   +---------------------------------+
                    ^
                    |
+-----------------------------------------+
|             <<interface>>               |
|  java.util.concurrent.ExecutorService   |
+-----------------------------------------+
| +shutdown(): void                       |
| +shutdownNow(): List<Runnable>          |
| +isShutdown(): boolean                  |
| +isTerminated(): boolean                |
| +newFixedThreadPool(numberOfThreads:    |
|  int): ExecutorService                  |
| +newCachedThreadPool(): ExecutorService |
+-----------------------------------------+

import java.util.concurrent.*;

public class ExecutorDemo {
	public static void main(String[] args) {
    	// 최대 3개의 고정된 thread pool 생성
    	ExecutorService executor = Executors.newFixedThreadPool(3);
        
        //executor에 runnable 객체 제출
        executor.execute(new PrintChar('a', 100));
        executor.execute(new PrintChar('b', 100));
        executor.execute(new PrintNum(100));
        
        //executor 닫기
        executor.shutdown();
    }
}

Thread pool
thread들을 모아 놓은 대기소. 작업이 들어올 때마다 thread를 새로 만드는 것이 아닌, 미리 일정 개수의 thread를 만들어놓고 들어오는 작업을 이 thread들이 나눠서 처리한다.
위 코드같은 경우는 3개의 thread를 미리 만들어놓은 것이다.

만약 위 코드에서 newFixedThreadPool(1);로 설정한다면 제출된 세 개의 Runnable 객체는 순차적으로 실행될 것이다. 풀에 있는 thread가 한 개이기 때문이다.

만약 위 코드에서 newCachedThreadPool();로 설정한다면 매 작업마다 새로운 thread가 생성될 것이다. 따라서 작업들은 동시에 실행될 것이다.

shutdown() 메소드는 말 그대로 executor를 닫아 새로운 task가 executor에 등록될 수 없도록 한다. 그러나, 이미 등록된 task들은 작업을 완수한다.

만약 하나의 task를 위해 한 개의 thread가 필요하다면 Thread class를, 여러 task를 위한 여러 개의 thread가 필요하다면 thread pool을 사용하는 것을 추천한다.

Thread Synchronization

의존성이 있는 thread의 실행을 조정하는 작업이다.
공유 리소스에 여러 thread가 동시에 접근할 경우 데이터가 손상될 수 있다.
당신이 100개의 thread를 생성하고 실행했다고 해보자. 각 thread들은 계좌에 돈을 입금하는 작업을 수행할 것이다. 계좌의 역할을 수행할 Account 클래스, Account에 돈을 넣기 위해 AddAPennyTask 클래스, 그것들을 생성하고 실행할 main 클래스를 정의한다.
이 클래스들의 관계는 아래와 같다.

+---------------+100   1+-----------------------------+
| AddAPennyTask |-----◆|     AccountWithoutSync      |
+---------------+       +-----------------------------+
| +run(): void  |       | -account: Account           |
+---------------+       | +main(args: String[]): void |
                        +-----------------------------+
                                      ◆1
                                       |
                                       | 1
                        +-----------------------------+
                        | Account                     |
                        +-----------------------------+
                        | -balance: int               |
                        | +getBalance(): int          |
                        | +deposit(amount: int): void |
                        +-----------------------------+

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

public class Main {
  private static Account account = new Account();

  public static void main(String[] args) {
    try (ExecutorService executor = Executors.newCachedThreadPool()) {

      // Create and launch 100 threads
      for (int i = 0; i < 100; i++) {
        executor.execute(new AddAPennyTask());
      }

      executor.shutdown();

      while (!executor.isTerminated()) {}
    }

    System.out.println("Balance: " + account.getBalance());
  }

  private static class AddAPennyTask implements Runnable {

    @Override
    public void run() {
      account.deposit(1);
    }
  }

  private static class Account {
    private int balance = 0;

    public int getBalance() {
      return balance;
    }

    public void deposit(int amount) {
      int newBalance = balance + amount;
      try {
        Thread.sleep(5);
      } catch(InterruptedException ex) {
      }
      balance = newBalance;
    }
  }
}

예상대로라면 최종 결과는 100페니가 출력되어야 한다.
그러나, 결과는 아래와 같다. (결과는 실행 마다 바뀔 수 있음)

Balance: 2

문제 원인은 여기에 있다.

    public void deposit(int amount) {
      int newBalance = balance + amount; // 여기와
      try {
        Thread.sleep(5);
      } catch(InterruptedException ex) {
      }
      balance = newBalance; // 여기
    }
  }

이 코드는 의도적으로 data corruption 문제가 발생하도록 작성되었다.
이 프로그램에서 오류가 발생하는 시나리오는 아래와 같다.

Step	Balance	Task 1	Task 2
1	0	newBalance = balance + 1;
2	0		newBalance = balance + 1;
3	1	balance = newBalance;
4	1		balance = newBalance;

이러한 경쟁 조건(Race condition)을 피하기 위해 2개 이상의 thread가 동시에 프로그램의 중요한 부분(critical region)에 들어오는 것을 반드시 막아야한다.

방법 1: synchronized
위 코드에서 critical region은 deposit 메소드이다.
synchronized 키워드를 사용하면 동시에 하나의 thread만 메소드에 접근할 수 있도록 만들 수 있다.

Step	Balance	Task 1	Task 2
1	0	newBalance = balance + 1;	wait...
2	1	balance = newBalance;	wait...
3	1		newBalance = balance + 1;
4	2		balance = newBalance;

객체의 synchronized 인스턴스 메소드를 호출하여 객체를 lock할 수 있다.
class의 synchronized static 메소드를 호출하여 class를 lock할 수 있다.
synchronized 구문은 this 객체 뿐만 아니라 어느 객체든 잠글 수 있다.

synchronized (expr) {
	statements;
}

이 구문을 위 코드에 적용하면 아래와 같다.

// 객체에 synchronized
synchronized (account) {
	account.deposit(1);
}

// 메소드에 synchronized
public synchronized void deposit(double amount) {
}

shnchronized 인스턴스 메소드는 실행되기 전에 명시적으로 인스턴스에 lock을 부여한다.

방법 2: Lock
lock은 Lock 인터페이스의 인스턴스이다. 또한 newCondition() 메소드를 사용해 thread 소통에 사용되는 Condition 객체를 만들 수 있다.

   +---------------------------------+
   |          <<interface>>          |
   | java.util.concurrent.locks.Lock |
   +---------------------------------+
   | +lock(): void                   |
   | +unlock(): void                 |
   | +newCondition(): Condition      |
   +---------------------------------+
                    ^
                    |
+------------------------------------------+
| java.util.concurrent.locks.ReentrantLock |
+------------------------------------------+
| +ReentrantLock()                         |
| +ReentrantLock(fair: boolean)            |
+------------------------------------------+

아래 코드에서 Lock 객체의 활용을 확인하자.

import java.lang.reflect.Executable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Main {
  private static Account account = new Account();

  public static void main(String[] args) {
    try (ExecutorService executor = Executors.newCachedThreadPool()) {

      for (int i=0; i<100; i++) {
        executor.execute(new AddAPennyTask());
      }

      executor.shutdown();

      while (!executor.isTerminated()) {}

      System.out.println("Balance: " + account.getBalance());
    }
  }

  public static class AddAPennyTask implements Runnable {

    @Override
    public void run() {
      account.deposit(1);
    }
  }

  public static class Account {
    private static Lock lock = new ReentrantLock(); // Lock 생성
    private int balance = 0;

    public int getBalance() {
      return balance;
    }

    public void deposit(int amount) {
      lock.lock(); // 명시적으로 잠금
      try {
        int newBalance = balance + amount;
        Thread.sleep(5);
        balance = newBalance;
      } catch (InterruptedException ex) {

      } finally {
        lock.unlock(); // 명시적으로 잠금 해제
      }
    }
  }
}

이 코드의 실행 결과는 처음 우리가 예상했던 것과 같다.

Balance: 100

lock을 try-catch 블럭과 함께 호출하고, finally에서 lock을 해제하는 이러한 코드 작성법을 권장한다.
(데드락 차단)

---

<참고자료>
탁성우, "플랫폼기반프로그래밍", 부산대학교