수면제가 필요해요...

package study.concurrency.tmp;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class AwaitTerminationTest {

 ExecutorService exec       = Executors.newCachedThreadPool();
 final Queue<String> resultQueue    = new ConcurrentLinkedQueue<String>();
 
 public void test() throws InterruptedException {
  for(int i = 0;i<10;i++)
   exec.execute(new WorkerTask(resultQueue));
  
  exec.shutdown();
 
  // 설정한 시간까지 모든 작업이 종료되기를 기다린다.
  // 설정한 시간이 지나면 해당 스레드는 interrupted 된다.
  boolean isTeminated = exec.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
 
  // 해당 작업이 종료된 후 출력된다.
  System.out.println("is teminated : " + isTeminated);
 
  for(String s : resultQueue) {
   System.out.println(resultQueue.poll());
  }
 
 }
 
 public static void main(String[] args) {
  try {
   new AwaitTerminationTest().test();
  } catch (InterruptedException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
}

package study.concurrency.tmp;

import java.util.Collection;

public class WorkerTask implements Runnable {
 
 final Collection<String> resultQueue;
 
 WorkerTask(final Collection<String> resultQueue) {
  this.resultQueue = resultQueue;
 }
 
 public void run() {
  // TODO Auto-generated method stub
  for(int i = 0;i<5;i++) { 
   resultQueue.add(Thread.currentThread().getName() + " : ["+ i +"]");
   try {
    Thread.sleep(500);
   } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
   }
  }
 }

}    

package java.util.concurrent.locks;

import java.util.concurrent.*;
import java.util.Date;

public interface Condition {
    void await() throws InterruptedException;
    void awaitUninterruptibly();
    long awaitNanos(long nanosTimeout) throws InterruptedException;
    boolean await(long time, TimeUnit unit) throws InterruptedException;
    boolean awaitUntil(Date deadline) throws InterruptedException;
    void signal();
    void signalAll();
}

package study.lock.condition;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

class BoundedBuffer<T> {
 final Lock   lock   = new ReentrantLock();
 final Condition notFull  = lock.newCondition();
 final Condition notEmpty = lock.newCondition();
 final T[]   items   = (T[]) new Object[100];
 int tail, head, count;

 public void put(T x) throws InterruptedException {
  lock.lock();
  try {
   while (count == items.length)
    notFull.await();
   items[tail] = x;
   if (++tail == items.length) tail = 0;
   ++count;
   notEmpty.signal();
  } finally {
   lock.unlock();
  }
 }

 public Object take() throws InterruptedException {
  lock.lock();
  try {
   while (count == 0)
    notEmpty.await();
   Object x = items[head];
   if (++head == items.length) head = 0;
   --count;
   notFull.signal();
   return x;
  } finally {
   lock.unlock();
  }
 }
}

package study.lock;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class LockSample {

 final Lock lock = new ReentrantLock();
 
 public static Thread thread;
 
 public void lockTest() throws InterruptedException {
 
  if(!lock.tryLock(1, TimeUnit.SECONDS)) {
   System.out.println(Thread.currentThread().getName() + " : 누군가가 선점하고 있다.");
   return;
  }
 
  thread = Thread.currentThread();
 
  System.out.println(Thread.currentThread().getName() + " : 내가 선점하고 있다.");
 
  try {
     // 락을 인터럽트 시킬수 있다.
     lock.lockInterruptibly();
    
   try {
      // critical section
    Thread.sleep(6000);
     } finally {
      // 획 해제
      lock.unlock();
     }
   } catch (InterruptedException e) {
    // 인터럽트 됨.
  
    throw new InterruptedException(Thread.currentThread().getName() + " : 인터럽트 당함");
   }  finally {
    lock.unlock();
   }
 }
}

package study.lock;


public class Task extends Thread {
 
 final LockSample sample;
 
 public Task(String name, final LockSample sample) {
  super(name);
  this.sample = sample;
 }
 
 public void run() {

  while(!isInterrupted())  {
   try {

    sample.lockTest();
   } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    return;
   }
  }
 }
}

final LockSample sample = new LockSample();
 
Task task = new Task("task1", sample);
task.start();

Task task1 = new Task("task2", sample);
task1.start();

try {
 Thread.sleep(5000);
} catch (InterruptedException e) {
 // TODO Auto-generated catch block
 e.printStackTrace();
}

LockSample.thread.interrupt();

task2 : 내가 선점하고 있다.
task1 : 누군가가 선점하고 있다.
task1 : 누군가가 선점하고 있다.
task1 : 누군가가 선점하고 있다.
task1 : 누군가가 선점하고 있다.
task1 : 내가 선점하고 있다.
java.lang.InterruptedException: task2 : 인터럽트 당함
task1 : 내가 선점하고 있다.
task1 : 내가 선점하고 있다.
task1 : 내가 선점하고 있다.
task1 : 내가 선점하고 있다.
task1 : 내가 선점하고 있다.

http://blog.sdnkorea.com/blog/178?category=15
java concurrency in practice

암묵적인 락 기능이 있는데 왜 명시적인 락 클래스를 따로 만들었을까? 암묵적인 락(synchronized)로 수용할 수 없는 문제(기능적으로 제한되는 경우)가 발생된다.

ex> 락을 확보하고자 대기하고 있는 쓰레드에 인터럽트를 걸거나, 대기 상태에 들어가지 않으면서 락을 확보하는 방법

성능
- 자바 5.0 에서는 ReentrantLock 은 암묵적인 락에 비해 훨씬 낳은 경쟁 성능(contended performance)을 보여준다.

자바 6 의 암묵적인 락은 개선되어 좀더 낳은 성능을 보임

Lock lock = new ReentrantLock();
// 락 획득
lock.lock();
try {
 // critical section
} finally {
 // 획 해제
 lock.unlock();
}

Lock lock = new ReentrantLock(); 

// 락을 인터럽트 시킬수 있다.
lock.lockInterruptibly();

try {
    // critical section
} finally {
    // 획 해제
    lock.unlock();
}

Lock lock = new ReentrantLock(); 

if(lock.tryLock()) return;

try {
    // critical section
} finally {
    // 획 해제
    lock.unlock();
}

ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
Lock readLock  = readWriteLock.readLock();
Lock writeLock = readWriteLock.writeLock();

package study.concurrency.invoke;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class InvokeAllTest {

 final ExecutorService executorService  = Executors.newFixedThreadPool(5);
 final List<WorkerTask> list    = new ArrayList<WorkerTask>();
 
 public void makeTaskList() {
  for(int i = 0;i<10;i++) {
   list.add(new WorkerTask());
  }
 }
 
 public void test() {
  try { 
   List<Future> futureList = executorService.invokeAll((Collection)list);
  
   try {
    for(Future future : futureList ) {
     System.out.println(future.get());
    }
   } catch (ExecutionException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
   }
   executorService.shutdown();

  } catch (InterruptedException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }

  System.out.println("Completed");
 }
 
 public static void main(String[] args) {
  new Launcher().test();
 }
}

package study.concurrency.invoke;

import java.util.concurrent.Callable;

public class WorkerTask implements Callable<String> {

 public String call() throws Exception {
  // TODO Auto-generated method stub
  for(int i = 0;i<5;i++) {
   System.out.println(Thread.currentThread().getName() + ":" + "running...");
   Thread.sleep(500);
  }
  return Thread.currentThread().getName() + ":" + "success";
 }
}

package study.threadpool.server;

import java.io.IOException;


import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class SimpleServer {

 // 스레드 풀은 실행할 작업이 없어도 corePoolSize 를 유지한다.
 // 최초에 corePoolSize 만큼 스레드를 생성하지 않고 작업이 실행 될 때 corePoolSize 만큼 차례대로 생성함.
 // prestartAllCoreThreads() 를 사용하면 corePoolSize 만큼 스레드를 미리 생성함.
 final int corePoolSize   = 10;
 // xaximumPoolsize 는 동시에 얼마나 많은 개수의 스레드가 동작할 수 있는지를 제한하는 최대값.
 final int xaximumPoolsize  = 10;
 final int blocdingQueueSize = 100;
 final int port    = 10001;
 // keepAliveTime 는 스레드 유지 시간으로 스레드가 keepAliveTime 이상 대기하고 있으면
 // 해당 스레드는 제거 될 수 있음. 풀의 스레드 개수가 corePoolSize 를 넘어서면 제거될 수 있음.
 final long keepAliveTime = 30L;

 final ThreadPoolExecutor pool;
 final ServerSocket servrerSocket;
 
 final BlockingQueue<Runnable> queue;
 final PoolThreadFactory threadFactory;
 
 public SimpleServer() throws IOException {
  queue    = new ArrayBlockingQueue<Runnable>(blocdingQueueSize);
  threadFactory  = new PoolThreadFactory("SERVER_POOL");
  servrerSocket  = new ServerSocket(port);
  pool    = new ThreadPoolExecutor(corePoolSize,
             xaximumPoolsize,
             keepAliveTime,
             TimeUnit.SECONDS,
             queue,
             threadFactory,
             new ThreadPoolExecutor.CallerRunsPolicy());
 
 }
 
 public void serverStart() {
  try {

   while(true) {
    Socket clientSocket = servrerSocket.accept(); 
    pool.execute(new Work(clientSocket));
   } 
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
 
 public static void main(String[] args) {
  try {
   new SimpleServer().serverStart();
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
}

package study.threadpool.server;

import java.util.concurrent.ThreadFactory;

public class PoolThreadFactory implements ThreadFactory {

 private final String name;
 
 public PoolThreadFactory(String name) {
  this.name = name;
 }
 
 @Override
 public Thread newThread(final Runnable r) {
  final PoolAppThread poolAppThread = new PoolAppThread(r, name);
  PoolAppThread.setDebug(true);   
  return poolAppThread;
 }
}

package study.threadpool.server;

import java.util.concurrent.atomic.AtomicInteger;

public class PoolAppThread extends Thread {

 public  static final String DEFAULT_POOL_NAME  = "PoolAppThread";
 private static volatile boolean debugLifecycle  = false;
 private static final AtomicInteger created   = new AtomicInteger();
 private static final AtomicInteger alive   = new AtomicInteger();
 
 public PoolAppThread(Runnable r) {
  this(r, DEFAULT_POOL_NAME); 
 }
 
 public PoolAppThread(Runnable runnable, String name) {
        super(runnable, name + "-" + created.incrementAndGet());
        setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
            public void uncaughtException(Thread t,
                                          Throwable e) {
            
             System.out.println("UNCAUGHT in thread " + t.getName() + ":" + e);
            }
        });
    }
 
 public void run() {
        // Copy debug flag to ensure consistent value throughout.
        boolean debug = debugLifecycle;
        if (debug) {
         System.out.println("Created : " + getName());        
        }
        try {
            alive.incrementAndGet();
            super.run();
        } finally {
            alive.decrementAndGet();
            if (debug) {
             System.out.println("Exiting : " + getName());
            }
        }
    }
 
 public static int getThreadsCreated() {
        return created.get();
    }

    public static int getThreadsAlive() {
        return alive.get();
    }

    public static boolean getDebug() {
        return debugLifecycle;
    }

    public static void setDebug(boolean b) {
        debugLifecycle = b;
    }
}

package study.threadpool.server;

import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;


public class Work implements Runnable {

 
 Socket socket;
 
 public Work(Socket socket) {
  this.socket = socket;
 }
 
 @Override
 public void run() {
  // TODO Auto-generated method stub
  try {
   PrintWriter out = new PrintWriter(socket.getOutputStream(), true);
   InputStreamReader in = new InputStreamReader(socket.getInputStream());
  
   char[] rBuf = new char[1000];
   int readMsg = in.read(rBuf);
  
   String msg = String.valueOf(rBuf).trim();
  
   System.out.println("read : [" + msg + "]");
  
   try {
    Thread.sleep(50);
   } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
   }
  
   out.println("이거나 받으삼\n");
  
  } catch (IOException e) {
   // TODO Auto-generated catch block
   System.err.println("Oh, dear me! " + e);
  }
 }
}

package study.threadpool.timing;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;

public class TimingThreadPoolExecutor extends ThreadPoolExecutor {

    public TimingThreadPoolExecutor(int corePoolSize,
            int maximumPoolSize,
            long keepAliveTime,
            TimeUnit unit,
            BlockingQueue<Runnable> workQueue) {
  super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
 }

 private final ThreadLocal<Long> startTime = new ThreadLocal<Long>();
    private final AtomicLong numTasks = new AtomicLong();
    private final AtomicLong totalTime = new AtomicLong();

    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        startTime.set(System.nanoTime());
    }

    protected synchronized void afterExecute(Runnable r, Throwable t) {
        try {
            long endTime = System.nanoTime();
            long taskTime = endTime - startTime.get();
            numTasks.incrementAndGet();
            totalTime.addAndGet(taskTime);
        } finally {
            super.afterExecute(r, t);
        }
    }

    protected synchronized void terminated() {
        try {
         System.out.println("Terminated:");
         System.out.println(String.format(" total    time=%dns", totalTime.get()));
         System.out.println(String.format(" numTasks time=%d", numTasks.get()));        
            System.out.println(String.format(" avg time=%dns", (totalTime.get() / numTasks.get())));
                    
        } finally {
            super.terminated();
        }
    }
}

package study.threadpool.timing;

public class Work implements Runnable {

 @Override
 public void run() {
  // TODO Auto-generated method stub
  try {
   for(int i = 0;i<3;i++) {
    System.out.println("running...[" + i + "]");
    Thread.sleep(1000);
   }
  } catch (InterruptedException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
}

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Launcher {

 final ThreadPoolExecutor pool;
 final BlockingQueue<Runnable> queue;

 public Launcher() {
  queue    = new ArrayBlockingQueue<Runnable>(1);
  pool    = new TimingThreadPoolExecutor(1, 1, 30, TimeUnit.SECONDS, queue);
 }
 
 public void test() {
  for(int i = 0;i<2;i++)
   pool.execute(new Work());
 
  pool.shutdown();
 }
 public static void main(String[] args) {
  new Launcher().test();
 }
}

running...[0]
running...[1]
running...[2]
running...[0]
running...[1]
running...[2]
Terminated:
 total    time=5999990250ns
 numTasks time=2
 avg time=2999995125ns

package study.shutdownhook;

public class Launcher {

 public static void main(String[] args) {
  Runtime.getRuntime().addShutdownHook(new Thread() {
   public void run() {
    System.out.println("종료훅 실행..");
   }
  });
 
  ShutdownHookThread t = new ShutdownHookThread();
  t.start();
 
 
  try {
   Thread.sleep(5000);
  } catch (InterruptedException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
  t.interrupt();
 }
}

class ShutdownHookThread extends Thread {
 public void run() {
  try {
   while(!Thread.currentThread().isInterrupted()) {
    System.out.println("running...");
  
    Thread.sleep(1000);
   }
  } catch (InterruptedException ignored) {}
 }
}

package study.interrupt.future;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class Launcher {

 private static final ExecutorService exce = Executors.newSingleThreadExecutor();
 
 public static void main(String[] args) {
  BlockingQueue<String> queue = new ArrayBlockingQueue<String>(10);
 
  Future<?> task = exce.submit(new Task(queue));
  
  try {
   task.get(3, TimeUnit.SECONDS);
  } catch (InterruptedException e) {
   e.printStackTrace();
  } catch (ExecutionException e) {
   e.printStackTrace();
  } catch (TimeoutException e) {
   e.printStackTrace();
   task.cancel(true);
  }

  if(task.isCancelled()) {
   System.out.println("종료됨");
  
  }
  for(String str : queue) {
   System.out.print(str + " : " );
  }

 }
}

package study.interrupt.future;

import java.util.concurrent.BlockingQueue;

public class Task implements Runnable {

 private final BlockingQueue<String> queue;
 
 Task(BlockingQueue<String> queue) {
  this.queue = queue;
 }
 
 @Override
 public void run() {
  try {
  
   int index = 0;
  
   while(!Thread.currentThread().isInterrupted()) {
    queue.put("" + index);
    System.out.print(index + " : ");
    index ++;
  
    Thread.sleep(50);
   }
  } catch(InterruptedException e) {
   e.printStackTrace();
  }
 }
}

package study.interrupt;

import java.util.concurrent.BlockingQueue;

public class SampleThread extends Thread {

 private final BlockingQueue<String> queue;
 
 SampleThread(BlockingQueue<String> queue) {
  this.queue = queue;
 }
 
 @Override
 public void run() {
  try {
  
   int index = 0;
  
   while(!Thread.currentThread().isInterrupted()) {
    queue.put("" + index);
    System.out.print(index + " : ");
    index ++;
   }
  } catch(InterruptedException e) {
   System.out.println();
  }
 }
 
 public void cancel() {
  interrupt();
 }

}

package study.interrupt;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class Launcher {

 public static void main(String[] args) {
  BlockingQueue<String> queue = new ArrayBlockingQueue(10);
 
  SampleThread t = new SampleThread(queue);
  t.start();
 
  try {
   Thread.sleep(3000);
  } catch (InterruptedException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
  t.cancel();

  for(String str : queue) {
   System.out.print(str + " : " );
  }

 }
}