HAMA 블로그

소프트웨어 엔지니어링에서 풀의 종류는 다양한데요.

쓰레드풀,메모리풀,캐쉬풀,커넥션풀,객체풀 (자바에서 객체풀은 사용을 지양합니다. 메모리를 할당하는 작업이 C/C++보다 빠름) 등등이 있습니다. "풀"어서 말하면 미리 만들어두고 돌려막기로 사용하자 라고 볼 수 있는데요. 미리 만들어 두는 방식 / 쓰레드가 태스크를 처리하는 방식에 따라서 다양한 풀의 구현체들이 있을 수 있습니다.  이 글에서는 openJDK8 기준의 자바에서 구현된 newFixedThreadPool 를 해부해보도록 하겠습니다.  

1.newFixedThreadPool 사용법 (자바8기준)

execute 메소드

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream; 
public class SimpleThreadPool {
  public static void main(String[] args) {   
  ExecutorService executor = Executors.newFixedThreadPool(10);   
  IntStream.range(0, 10).forEach( n -> executor.execute( () -> 
        {       
          try {         
                TimeUnit.MILLISECONDS.sleep(300);         
                String threadName = Thread.currentThread().getName();         
                System.out.println("Hello " + threadName);        
          } catch (InterruptedException e) {
             e.printStackTrace();        
          }     
      })   
  );  
}
}

- 쓰레드풀에 쓰레드를 10개를 뛰어놀게 한다. 
- 10번을 반복해서 쓰레드풀에 일을 시킨다. (쓰레드풀안의 쓰레드 하나가 선택되어 일처리를 할것이다)
- execute 메소드를 사용했다. 이 메소드는 void 를 리턴한다. 즉 일처리를 시키기만 하지 결과를 보고받지 않을것이다.
- 구현내용은 300밀리초를 기다렸다가 Hello 쓰레드이름을 출력하는 것이다. 

- 자바8부터는 for(int i = 0; i < 10; i++) 보다는 저렇게 사용하는게 좋다.

submit 메소드

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;

public class SimpleThreadPool {

  public static void main(String[] args) {

    ExecutorService executor = Executors.newFixedThreadPool(2);
    final List<Integer> integers = Arrays.asList(1,2,3,4,5);
    Future<Integer> future = executor.submit(() -> {
      TimeUnit.MILLISECONDS.sleep(5000);
      int result = integers.stream().mapToInt(i -> i.intValue()).sum();
      return result;
    });

    try {
      Integer result = future.get();
      System.out.print("result: " + result);
      executor.shutdownNow();
    } catch (InterruptedException e) {
      e.printStackTrace();
    } catch (ExecutionException e) {
      e.printStackTrace();
    }
  }
}

- 쓰레드풀에 쓰레드를 10개를 뛰어놀게 한다. 
- 쓰레드풀에 하나의 일을 시킨다. (쓰레드풀안의 쓰레드 하나가 선택되어 일처리를 할것이다)
- submit 메소드를 사용했다. 이 메소드는 future 를 리턴한다. 즉 일처리를 시키기고, 그에 따른 결과를 보고받을 것이다.
- 구현내용은 5000밀리초를 기다렸다가 리스트안의 숫자들의 합을 리턴한다.

- 리턴 받은 future 로 부터 값을 얻는다. 여기서 get()메소드는 블럭된다. (타임아웃을 매개변수로 넣을 수도 있다) 아마 일처리를 하는 쪽의 쓰레드에서 일을 다 끝내고 set() 같은 것을 해 줄 때가지 블럭될거 같다.

이제 사용법은 알았으니 과연 쓰레드풀을 자바에서 어떻게 구현하고 있는지 확인 해보자.

@어떻게 10개의 쓰레드를 가진 풀을 만드는지? (newFixedThreadPool(10))
@어떻게 풀에서 쓰레드 하나를 할당해 주는지? (execute)

@쓰레드풀에서 할당 될 워커(쓰레드)가 없을 때 태스크가 어떻게 되는지?

@어떻게 future 를 리턴해주는지. (submit) 

2.newFixedThreadPool 소스 분석(자바8기준)

* 참고로 지면이 한정된 관계로 예외처리 포함해서 여러 로직들이 생략되었다.

(발췌: http://codepumpkin.com/threadpool-using-executor-framework/)

2-1. 어떻게 10개의 쓰레드를 가진 풀을 만드는지? (newFixedThreadPool(10))

public class Executors {

java/util/concurrent/Executors.java 라는 파일에는 Executors 라는 클래스가 있으며, 정적메소드로써 newFiexedThreadPool 이라는 것을 제공한다. 매개변수는 쓰레드풀에 들어갈 쓰레드들의 개수를 넣어준다.

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>());
}

퍼사드 역할 (Gof의 패턴중 하나로 내부 시스템의 복잡함을 감추고 사용자가 간편히 사용하도록 하는 의도)

을 하고있다. 그저 팩토리로써 내부의 ThreadPoolExecutor 객체를 대신 만들어주고 있는데, 매개변수로는 순서대로 살펴보자.

- corePoolSize : 풀 안에 유지되는 쓰레드 개수 (시작시) 

- maximumPoolSize : 풀에 유지되는 최대 쓰레드 개수

- keepAliveTime : corePoolSize 보다 쓰레드 개수가 많아 질 때, 새로운 테스크를 기다리기 위한 시간. 시간이 지나면 쓰레드를 없애서 corePoolSize 를 유지한다.

- unit :  keepAliveTime 시간단위

- workQueue : 실행 되기전에 홀드시켜 두는 태스크를 유지하는 큐. 쓰레드가 남지 않을 경우 여기 태스크를 넣는다.

시작개수가 1이고 최대개수가 1이면  newSingleThreadExecutor 이 되는것이고,

시작개수가 0이고 최대개수가 MAX_VALUE 가 되면, newCachedThreadPool 풀이라 한다.
우리는 최소,최대가 10개로 고정된 쓰레드풀을 만든 것이다.

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

java/util/concurrent/ThreadPoolExecutor.java 라는 파일에는 ThreadPoolExecutor라는 클래스가 있으며, 들어온 매개변수에 추가로 Executors.defaultThreadFactory()와 defaultHandler 를 추가 매개변수로 받아서 새 객체를 만든다.

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

 Executors.defaultThreadFactory()

public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
       
        return t;
    }
}

public void execute(Runnable command) {

  int c = ctl.get(); 
  // 1 단계 
  if (workerCountOf(c) < corePoolSize) {
    if (addWorker(command, true))
      return;

  }

  // 2 단계
    .... 생략 ....
}

private boolean addWorker(Runnable firstTask, boolean core) {
  .. 상태체크 부분 생략 .. 

  Worker w = null;

  w = new Worker(firstTask); // (1)
  final Thread t = w.thread;
  if (t != null) {

    workers.add(w); // (1)
    t.start();
    workerStarted = true;

  }

  return workerStarted;
}

  w = new Worker(firstTask); // (1)

Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

workers.add(w);

t.start();

public void execute(Runnable command) {
        // 1단계
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 2단계 
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

 final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
  
        while (task != null || (task = getTask()) != null) {

            ... 생략 ...

            task.run();
                   
            ... 생략 ...
        }
           
    }

 private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

        for (;;) {
                   .. 생략 ..

            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
              
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

public <T> Future<T> submit(Callable<T> task) {
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

public FutureTask(Callable<V> callable) {
  this.callable = callable;
  this.state = NEW; // ensure visibility of callable
}

public void execute(Runnable command) {

  int c = ctl.get(); //1 단계
  if (workerCountOf(c) < corePoolSize) {
    if (addWorker(command, true))
      return;
  c = ctl.get();
  }

  //2 단계
    .... 생략 ....
}

public void run() {
  runWorker(this);
}

final void runWorker(Worker w) {
  Thread wt = Thread.currentThread();
  Runnable task = w.firstTask;
  w.firstTask = null;

  try {

    task.run();

  } finally {
    processWorkerExit(w, completedAbruptly);
  }
}

  task.run();

public void run() {

  try {
    Callable<V> c = callable;
    if (c != null && state == NEW) {
    V result;
    boolean ran;
    try {
      result = c.call();
      ran = true;
    } catch (Throwable ex) {
      result = null;
      ran = false;
      setException(ex);
    }
    if (ran)
      set(result);
    }
  }

....
}

if (ran)
    set(result);

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();
    }
}

public V get() throws InterruptedException, ExecutionException {
  int s = state;
  if (s <= COMPLETING)
    s = awaitDone(false, 0L); // awaitDone 에서 블럭되고 있음을 알수 있다. 하염없이 기다림 
  return report(s);
}

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));  // RUNNING 상태와 쓰레드 개수1로 초기화

private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;  // 00011111 11111111 11111111 11111111 

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS; // 11100000 00000000 00000000 00000000
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

private static int ctlOf(int rs, int wc) { return rs | wc; }  // 2가지 상태를 하나의 Int 로 합쳐주는 함수

private static final int COUNT_BITS = Integer.SIZE - 3;   // 32-3 = 29  

private static final int CAPACITY   = (1 << COUNT_BITS) - 1;  // 00011111 11111111 11111111 11111111 

private static int workerCountOf(int c)  { return c & CAPACITY; }   // 현재 워커 갯수는? 

int c = ctl.get();   
int ws = workerCountOf(c);  // ws 는 0 

참고:

http://tutorials.jenkov.com/java-util-concurrent/executorservice.html 

http://hg.openjdk.java.net/jdk8/jdk8/jdk/file/00cd9dc3c2b5/src/share/classes/java/util/concurrent/Executors.java 

잡설

개인적으로 리소스 해제와 관련해서 가장 먼저 떠오르는것은 C++의 포인터이다. 리소스를 해제하지 않아서 생기는 문제 또는 해제한 리소스를 사용하려고 해서 생기는 문제는 대규모 솔루션에서는 가장 골치 아픈 걱정거리가 되곤 하는데 스택트레이스에 잡히지 않는 머나먼 곳에서 해당 포인터를 조작한 것들이기에 전체 코드를 샅샅히 살피지 않으면 풀기 힘든 문제로 남기도 한다. (널포인터 문제와 레이스 컨디션 문제는 모든 솔루션에서 가장 대다수의 버그로 리포팅되며, 다른 폭탄을 야기하며, 해결하기 어려운 2가지 문제) C++은 이를 해결하고자 다음과 같은 스마트 포인터들을 제공하지만 스마트포인터 종류 분석  제대로 잘 활용하는 것도 문제이다.

자바(VM류)가 떠오르면서 리소스 관리로 부터 어느정도는 해방시켜 주었다. 하지만 외부리소스(파일,소켓,디비connection 등 의 리소스) 에 관해서는 자바도 사용자가 직접 리소스해제를 해야하는데, 리소스해제를 까먹는 경우가 생겨나서 말썽이 된다. 개발자 잘못이라고? 뭐 그렇게 볼수도 있지만 바야흐로 2018년. 개발자가 잘못하지 않게 만들어주는 덕목도 중요하리라~

지금부터 소개할 내용은 자바와 스칼라에서의 자동 리소스 해제 및 비슷한 상황으로 볼 수 있는 동기화객체 자동 해제 에 관련된 것들이며 소개차원의 범위를 가진다. 또한 정답도 없다.

* 참고로 다른 언어들도 관련된 것들이 있다. 예를들어 파이썬의 with 구문

- ARM: Automatic Resource Management 자동 리소스 관리를 지칭하는 일반적인 용어
- EAM: Eexcute Around Method Pattern 함수 실행시 규칙적으로 일어나는 일을 묶어 놓는 패턴 (리소스 할당/해제에만 국한된 개념이 아니다. 트랜잭션,락등에도 사용할 수 있다. AOP ,Proxy Pattern, Decorator Pattern 등도 비슷한 일을 의미/활용하긴 하는데 이것들의 의미는 좀 더 넓다)  

 리소스릭 막기

Java 

다음 코드를 보자.

private void incorrectWriting() throws IOException {
  DataOutputStream out = new DataOutputStream(new FileOutputStream("data"));
  out.writeInt(666);
  out.writeUTF("Hello");
  out.close();
}

파일 리소스를 얻어서 데이터를 쓰고(write), flush 로 밀어내고,리소스를 반납(close)를 하는 소스이다. 여기서의 문제는

1. close를 잊어버리는 개발자가 많이 있다. (개발자 수준이 낮아서 그렇다고? 쩌는 개발자들이 참여하는 훌륭한 오픈소스에서도 equals 와 == 실수 조차도 엄청 많이 발견된다. 실수는 누구나 할 수 있다.) 

2. close 를 하더라도 쓰는 과정에서 예외가 발생할 경우 리소스 누수가 생겨난다. 

따라서 리소스 누수 문제를 해결하기 위해서 아래와 같이 코딩한다.

private void correctWriting() throws IOException {
  DataOutputStream out = null;
  try {
    out = new DataOutputStream(new FileOutputStream("data"));
    out.writeInt(666);
    out.writeUTF("Hello");
  } finally {
    if (out != null) {
      out.close();
    }
  }
}

보시다사피 try ~ finally 블럭으로 감싸서, 무슨일이 있어도 finally 블록안의 내용을 실행하도록 하였다.
문제는 없다. 하지만 먼가 상투적인 코드가 남발되어 있다는 느낌적인 느낌? 강제가 아니라서 대충 안하고 넘겨 버릴거 같은 느낌? 

Java7에서는 이렇게 해결한다.

Java7 (try-with-resources 제공)

private void writingWithARM() throws IOException {
  try (DataOutputStream out = new DataOutputStream(new FileOutputStream("data"))) {
    out.writeInt(666);
    out.writeUTF("Hello");
  }
}

별거 없다. 그냥 코드를 좀 줄여주는 장치를 추가한거다. try 블럭의 모양새가 변했다. 
try 문안에 해제되야할 리소스 할당부분을 적어주고, 본문에 로직을 적어 주면 된다. (참고로 python 은 with 문으로 할당부분을 감싼다) 그러면 자동적으로 try 문을 벗어날때 자동적으로 close 를 호출해 준다.
이 특별한 형식을 try-with-resources 라고 부른다.

근데 저런 형식이 거저로 얻어지는 것은 아니다. 클래스 만드는 사람은 손이 더 가야한다.
try-with-resources 를 사용하려면 java.lang.AutoCloseable 인터페이스를 구현해야한다.

public class AutoClose implements AutoCloseable {

  @Override
  public void close() {
    System.out.println(">>> close()");  // DataOutputStream 경우는 내부에서 스트림에 대한 close()를 호출

  }

  public void work() throws MyException {
    System.out.println(">>> work()");
  }
}

이렇게 AutoCloseable 인터페이스를 통해 close 를 오버라이드 하면 된다.

public static void main(String[] args) {
  try (AutoClose autoClose = new AutoClose()) {
    autoClose.work();
  } 
}

이제 try-with-resources 를 사용할 수 있다.

* 참고로 JDK7 의 OuputStream 를 보면 public abstract class OutputStream implements Closeable, Flushable { 로 되있는데 Closeable 은 AutoCloseable 을 상속받은 인터페이스이며 해당i/O 부분에 특화되어있다. (IOException 을 발생시킴) 

자바7 ARM 실용예제 실습 

import java.io.FileWriter;
import java.io.IOException;
import java.lang.AutoCloseable;

public class FileWriterARM implements AutoCloseable {
    private final FileWriter writer;

    public FileWriterARM(final String fileName) throws IOException {
        writer = new FileWriter(fileName);
    }

    public void writeStuff(final String message) throws IOException {
        writer.write(message);
    }

    public void close() throws IOException {
        System.out.println("close called automatically...");
        writer.close();
    }
    //...
    public static void main(final String[] args) throws IOException {
        try(final FileWriterARM writerARM = new FileWriterARM("peekaboo.txt")) {
                writerARM.writeStuff("peek-a-boo");
                System.out.println("done with the resource...");
        }
    }
}

뭐 좋은거 같긴하다. 근데 문제가 있을까? 저런게 (try 구문및 AutoCloseable 인터페이스등) 있다는 것을 기억하고 사용을 하게 사용자에게 책임을 넘겨야 한다는 것이 문제라고 하면 문제이다. 저걸 할 사람들이 모여있는 곳이라면 그냥 try ~finally 감싸고 close 도 잘 해주겠지. (농담이다. 저런것을 팀 차원에서 강제한다는거 자체가 문제를 해결하는 방법일 수도 있다.) 암튼 더 일반화/강제화 할 수 있는 방법도 있을까? Java8에서의 람다를 이용한 방식을 살펴보자. 

Java8 (EAM) 

Java8에는 드디어 염원하던 람다식이 도입되므로써, 함수형 프로그래밍을 자바에서 할 수 있는 멋진 도구를 선물 해 주었다.이 람다식을 활용하면 리소스 할당/해제라든지, 트랜잭션,락 시작/종료와 같은 '틀'에 박힌 행동을 다른 방식으로 관리 할 수 있게 해주는데 어떤식으로 하는지 살펴보자.  (다른 방식이지 정답이거나 가장 좋은 방식이라고 말 할 순 없다.) 

import java.util.function.Consumer;
import static java.lang.System.out;

class JavaResource {
  private JavaResource() {out.println("created...");}
  public void operation1() {out.println("operation 1");}
  public void operation2() {out.println("operation 2");}
  private void close() { out.println("cleanup");}

  public static void use(Consumer<JavaResource> block) {
    JavaResource resource = new JavaResource();
    try {
      block.accept(resource);
    } finally {
      resource.close();
    }
  }
}

Interface Consumer<T>가 있다. 자바8에서는 자주 이용되는 함수형 인터페이스를 java.util.function 을 통해서 제공하는데, 함수형 인터페이스는 메서드가 하나만 정의 되있다고 약속하였다. Consumer 인터페이스는 accept 를 통해 하나의 인자를 매개변수로 받아서 어떤 행위(객체T를 받아서 소비,부수효과를 일으킴)를 하고 리턴은 안하는 것으로, 함수형 프로그래밍에서는 최대한 부수효과(side effect) 를 없애지만 어쩔 수 없이 필요할때는 이렇게 잘보여지도록 표현한다. 

위의 코드에서는 객체생성,정리 부분을 자기가 처리하는게 아니라 use 메소드를 통해서 위임하고 있다. (private를 보라)  use 는 어떻게 보면 팩토리패턴(생성) + 리소스정리(close)가 함께 존재하는 것으로써, 내부에서 객체생성을 하고 close()메소드도 호출해준다.

(역시 자바는 accept 라는것을 강제하는등 조금 불편한 구석이 있다. 신규 개발은 스칼라로 해보는것도 좋을 것이다) 

public class Main {
  public static void main(String... args) {

    JavaResource.use(resource -> {
      resource.operation1();
      resource.operation2();
    });

  }
}

이제 생성,해제를 함께 해주는 JavaResource.use 팩토리에게 맞기고, 관심사인 부수효과적인 행위에 대한 것만 Interface Consumer<T>를 타겟으로 하는 람다식을 구현하면 된다. 

(resource -> {
      resource.operation1();
      resource.operation2();
    }

그 람다식은 위와 같으며, 위의 resource 는 accept 의 매개변수로 들어가서 operation1() 과 operation2() 라는 부수효과를 발생시킬 것이다. 결국 이런 정형화된 패턴의 코드를 통해 위험한(?) 객체를 직접적으로 사용하는 것을 금지당하므로써, 보다 안정적인 솔루션이 만들어 질 것이다.

좀 더 실용적인 예를 살펴보자.

람다식 + 함수형 인터페이스 실습 

import java.sql.Connection;
import java.sql.SQLException;

public interface Transaction {
  public void execute(Connection connection) throws SQLException;
}

Transaction 이라는 인터페이스가 있다. 한번의 트랜잭션을 일으키기 원할 때 사용하는데, 이 인터페이스를 어떻게 사용하는지 아래 예제에서 확인하자.

import java.sql.Connection;
import java.sql.DriverManager;

public class TransactionHandler {

  public static void runInTransaction(Transaction transaction) throws Exception {

    Connection dbConnection = createDatabaseConnection();
    dbConnection.setAutoCommit(false);

    try {

      System.out.println("Starting transaction");
      transaction.execute(dbConnection);

      System.out.println("Committing transaction");
      dbConnection.commit();

    } catch (Exception e) {

      System.out.println(e.getMessage());
      System.out.println("Rolling back...");
      dbConnection.rollback();
    } finally {
      dbConnection.close();
    }
  }

  private static Connection createDatabaseConnection() throws Exception {
    Class.forName("com.mysql.jdbc.Driver");
    return DriverManager.getConnection("jdbc:mysql://localhost:3306/ticket_system", "user", "password");
  }
}

코드 자체로 모든 설명이 들어가 있기에 추가적으로 설명 할 필요는 없을 거 같다.

"트랜잭션 핸들러" 클래스가 runInTransaction 이라는 스태틱 메소드를 통해 생성/정리를 대행해주고 있다. 전형적인 데코레이터 패턴이라고 볼 수 있다. (*자바에서는 이런곳에 사용하라고 전문적으로 다이내믹 프록시라는 기술을 제공하며, 하둡코어에서는 일반 함수를 호출하는 듯하지만 내부적으론 RPC 를 하는데 그것이 사용되고 있다.)

  public static void runInTransaction(Transaction transaction) throws Exception {

이렇게 Transaction 인터페이스에 해당하는 로직을 넘겨 줘야하는데, 

TransactionHandler.runInTransaction(connection -> {

  int ticketId = findAvailableTicket(connection);

  reserveTicket(ticketId, connection);
  markAsBought(ticketId, connection);
});

람다식을 통해 Transaction 인터페이스를 타겟으로 갖는 비지니스 로직을 가시권에 들어오게 직접 만들었다. 함수형 인터페이스는 메소드가 하나만 정의되 있으므로, execute 라는 메소드는 자동으로 매칭해준다.

람다식 + Consumer 함수형 인터페이스 실습 (리소스 해제)

import java.io.FileWriter;
import java.io.IOException;
import java.lang.AutoCloseable;

public class FileWriterEAM  {
  private final FileWriter writer;

  private FileWriterEAM(final String fileName) throws IOException {
    writer = new FileWriter(fileName);
  }
  private void close() throws IOException {
    System.out.println("close called automatically...");
    writer.close();
  }
  public void writeStuff(final String message) throws IOException {
    writer.write(message);
  }
  //...
  public static void use(final String fileName, final UseInstance<FileWriterEAM, IOException> block) throws IOException {

    final FileWriterEAM writerEAM = new FileWriterEAM(fileName);
    try {
      block.accept(writerEAM);
    } finally {
      writerEAM.close();
    }
  }

  public static void main(final String[] args) throws IOException {
    FileWriterEAM.use("eam.txt", writerEAM -> {
      writerEAM.writeStuff("how");
      writerEAM.writeStuff("sweet");
    });
  }
}

 use 를 통해서 객체생성,실행,해제를 모두 처리 해 주고 있다. 딱 정해져 있기 때문에 클래스 사용자가 실수를 범할 가능성이 없어졌다.

람다식 + Runnable 함수형 인터페이스 실습 (동기화 객체 자동 해제)

import java.util.concurrent.locks.Lock;

public class Locker {
  public static void runLocked(Lock lock, Runnable block) {
    lock.lock();

    try {
      block.run();
    } finally {
      lock.unlock();
    }    
  }
}

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import static fpij.Locker.runLocked;

public class Locking {
  Lock lock = new ReentrantLock(); //or mock
  protected void setLock(final Lock mock) {
    lock = mock;
  } 
  public void doOp1() {   // 행사코드 남발!!!!
    lock.lock();
    try {
      //...critical code...
    } finally {
      lock.unlock();
    }
  }
  
  // 깔끔해 졌다!!
  public void doOp2() {
    runLocked(lock, () -> {/*...critical code ... */});
  }
  public void doOp3() {
    runLocked(lock, () -> {/*...critical code ... */});
  }
  public void doOp4() {
    runLocked(lock, () -> {/*...critical code ... */});
  }
}

Scala (EAM) 

class ScalaResource private {
  println("created...")

  def operation1() = println("operation 1")
  def operation2() = println("operation 2")
  private def close() = println("cleaning up")
}
object ScalaResource {
  def use(closure : ScalaResource => Unit) = {
    val resource = new ScalaResource
    try {
      closure(resource)
    } finally {
      resource.close()
    }
  }
}
object MainEAM extends App {
  ScalaResource.use { resource =>
    resource.operation1()
    resource.operation2()
  }
}

-> 가 => 로 바뀌고 val, object같은 키워드가 있다는 것 정도의 기본 문법 차이를 제외하고 중요 포인트는 
- 스칼라는 굳이 Consumer 인터페이스라는것이 있을 필요가 없다는것~ 
- 자바에서 아주 군더더기 처럼 느껴졌던 accept 를 사용하지 않아도 된다.

참고:

http://www.oracle.com/technetwork/articles/java/trywithresources-401775.html 

https://www.geeksforgeeks.org/automatic-resource-management-java/ 

https://gist.github.com/dpsoft/9013481 

http://chrisoldwood.blogspot.kr/2011/07/execute-around-method-subsystem.html 

http://www.deadcoderising.com/transactions-using-execute-around-method-in-java-8/ 

https://pragprog.com/book/vsjava8/functional-programming-in-java