Java多线程并发编程

public class ThreadTest {
  public static void main(String[] args) {
    Thread thread = new ThreadDemo();
    thread.start();
  }
}

class ThreadDemo extends Thread {
  @Override
  public void run() {
    System.out.println("do something");
  }
}

public class ThreadTest {
  public static void main(String[] args) {
    Thread thread = new Thread(new RunnableDemo());
    thread.start();
  }
}

class RunnableDemo implements Runnable {
  @Override
  public void run() {
    System.out.println("do something");
  }
}

public class ThreadTest {
  public static void main(String[] args) throws Exception {
    FutureTask<String> futureTask = new FutureTask<>(new CallableDemo());
    Thread thread = new Thread(futureTask);
    thread.start();     //此时Callable接口已经被调用
    String result = futureTask.get();
    System.out.println(result);
  }
}

class CallableDemo implements Callable<String> {
  @Override
  public String call() throws Exception {
    return "do something";
  }
}

public class ThreadTest {
  public static void main(String[] args) {
    Timer timer = new Timer();
    timer.schedule(new TimerTaskDemo(),1000);
  }
}

class TimerTaskDemo extends TimerTask {
  @Override
  public void run() {
    System.out.println("do something");
  }
}

public class ThreadTest {
  public static void main(String[] args) {
    ExecutorService executor = Executors.newFixedThreadPool(5);
    executor.execute(new ExecutorDemo());
    executor.shutdown();	//线程池销毁，正在执行和在队列等待的线程不会销毁
  }
}

class ExecutorDemo implements Runnable {
  @Override
  public void run() {
    System.out.println("do something");
  }
}

public class DeadLockTest {

  public static final Object resource1 = new Object();
  public static final Object resource2 = new Object();

  public static void main(String[] args) {

    new Thread(() -> {
      synchronized (resource1) {
        try {
          Thread.sleep(500);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + "：等待");
        synchronized (resource2) {
          System.out.println(Thread.currentThread().getName() + "：可以执行");
        }
      }
    }, "A").start();

    new Thread(() -> {
      synchronized (resource2) {
        try {
          Thread.sleep(500);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + "：等待");
        synchronized (resource1) {
          System.out.println(Thread.currentThread().getName() + "：可以执行");
        }
      }
    }, "B").start();
  }

}

public class LiveLock {
  static volatile int count = 10;

  // 无法解锁，一个想加到20，一个想减到0
  public static void main(String[] args) {
    Thread t1 = new Thread(() -> {
      // 期望减到 0 退出循环
      while (count > 0) {
        try {
          TimeUnit.MILLISECONDS.sleep(500);
        } catch (InterruptedException e) {
          throw new RuntimeException(e);
        }

        count--;
        System.out.println("t1 count: " + count);
      }
    }, "t1");
    Thread t2 = new Thread(() -> {
      // 期望超过 20 退出循环
      while (count < 20) {
        try {
          TimeUnit.MILLISECONDS.sleep(500);
        } catch (InterruptedException e) {
          throw new RuntimeException(e);
        }

        count++;
        System.out.println("t2 count: " + count);
      }
    }, "t2");

    t1.start();
    t2.start();

    try {
      t1.join();
      t2.join();
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }

    System.out.println("main end");
  }
}

public class ThreadStarvingTest {

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

    executorService.submit(() -> {
      System.out.println(currentThreadName() + "接待客人");
      Future<String> cook = executorService.submit(() -> {
        System.out.println(currentThreadName() + "做回锅肉");
        return "回锅肉";
      });

      try {
        System.out.println(currentThreadName() + "上" + cook.get()); 
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (ExecutionException e) {
        e.printStackTrace();
      }
    });

    executorService.submit(() -> {
      System.out.println(currentThreadName() + "接待客人");
      Future<String> cook = executorService.submit(() -> {
        System.out.println(currentThreadName() + "做鱼香肉丝");
        return "鱼香肉丝";
      });

      try {
        System.out.println(currentThreadName() + "上" + cook.get());
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (ExecutionException e) {
        e.printStackTrace();
      }
    });

  }

  public static String currentThreadName() {
    return Thread.currentThread().getName();
  }

}

public class NotifyTest {

  public static void main(String[] args) throws InterruptedException {
    NotifyRunnable notifyRunnable = new NotifyRunnable();
    new Thread(notifyRunnable).start();
    TimeUnit.SECONDS.sleep(5);
    notifyRunnable.setFlag(false);
    synchronized (notifyRunnable) {
      notifyRunnable.notify();
    }
  }
}

class NotifyRunnable implements Runnable {
  private volatile boolean flag = true;

  public void setFlag(boolean flag) {
    this.flag = flag;
  }

  public synchronized void deal() {
    try {
      // synchronized的是当期对象，这里需要while循环，被唤醒之后是继续执行的
      while(flag) {
        wait();
      }
      System.out.println("唤醒了一个线程");
    } catch ( InterruptedException e) {
      e.printStackTrace();
    }
  }

  @Override
  public void run() {
    deal();
  }
}

public class Paker {

  public static void main(String[] args) {
    Thread t1  = new Thread(() -> {
      System.out.println("start...");

      try {
        TimeUnit.SECONDS.sleep(1);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }

      System.out.println("park...");
      LockSupport.park();
      System.out.println("resume...");
    });

    t1.start();
    try {
      TimeUnit.SECONDS.sleep(3);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }

    LockSupport.unpark(t1);
  }
}

public class InterruptTest {

  public static void main(String[] args) throws InterruptedException {
    Thread t1 = new Thread(()->{
      try {
        TimeUnit.SECONDS.sleep(3);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }, "t1");
    t1.start();
    TimeUnit.SECONDS.sleep(1);
    t1.interrupt();
    // 打印false
    System.out.println(" 打断状态: " + t1.isInterrupted());
  }
}

public class InterruptTest {

  public static void main(String[] args) throws InterruptedException {
    Thread t1 = new Thread(()->{
      while(true) {
        Thread current = Thread.currentThread();
        boolean interrupted = current.isInterrupted();
        if(interrupted) {
          System.out.println(" 打断状态: " + interrupted);
          break;
        }
      }
    }, "t2");
    t1.start();
    TimeUnit.SECONDS.sleep(1);
    t1.interrupt();
  }
}

public class InterruptTest {

  public static void main(String[] args) throws InterruptedException {
    Thread t1 = new Thread(() -> {
      System.out.println("park...");
      LockSupport.park();
      System.out.println("unpark...");
      // 打印 true
      System.out.println("打断状态： " + Thread.currentThread().isInterrupted());
    }, "t1");
    t1.start();
    TimeUnit.SECONDS.sleep(1);
    t1.interrupt();
  }

}

public class JoinDemo {

  public static void main(String[] args) {
    Thread t1 = new Thread(() -> {
      try {
        TimeUnit.SECONDS.sleep(3);
      } catch (InterruptedException e) {
        System.out.println(e);
      }
    });
    t1.start();

    try {
      t1.join();
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }

    // 3s 之后才会打印，即t1执行完成打印
    System.out.println("main end");
  }
}

public final void join() throws InterruptedException {
  join(0);
}

public final synchronized void join(long millis) throws InterruptedException {
  long base = System.currentTimeMillis();
  long now = 0;

  if (millis < 0) {
    throw new IllegalArgumentException("timeout value is negative");
  }

  if (millis == 0) {
    // 如果线程存活一直调用wait（即使被打断也会继续阻塞）
    while (isAlive()) {
      wait(0);
    }
  } else {
    // 有时间的等待，考虑了打断的影响
    while (isAlive()) {
      long delay = millis - now;
      if (delay <= 0) {
        break;
      }
      wait(delay);
      now = System.currentTimeMillis() - base;
    }
  }
}

public interface Lock {

  // 获取锁。如果锁不可用，将禁用当前线程，并且在获得锁之前，该线程将一直处于休眠状态
  void lock();

  // 获取锁，等待过程中如果当前线程被中断，则抛出异常，此时加锁还未成功不需要释放资源
  void lockInterruptibly() throws InterruptedException;

  // 仅在调用时锁为空闲状态才获取该锁。如果锁可用立即返回值true。如果锁不可用，立即返回值 false
  boolean tryLock();

  // 如果锁在给定的等待时间内空闲，并且当前线程未被中断，则获取锁，如果等待期间被中断，则抛出异常
  boolean tryLock(long time, TimeUnit unit) throws InterruptedException;

  // 释放锁。锁必须由当前线程持有。调用Condition.await()将在等待前以原子方式释放锁
  void unlock();

  // 返回绑定到此Lock实例的新Condition实例
  Condition newCondition();
}

//头节点
private transient volatile Node head;

//尾节点
private transient volatile Node tail;

//共享变量，使用volatile修饰保证线程可见性，用于记录了加锁次数
private volatile int state;

// 队列头节点，头结点不储存数据
private transient volatile Node head;

// 队列尾节点
private transient volatile Node tail;

static final class Node {
  /** 节点为共享模式下等待 */
  static final Node SHARED = new Node();

  /** 节点为独占模式下等待 */
  static final Node EXCLUSIVE = null;

  /** 当前节点由于超时或中断被取消 */
  static final int CANCELLED =  1;

  /** 表示当前节点的前节点被阻塞 */
  static final int SIGNAL    = -1;

  /** 当前节点在等待condition */
  static final int CONDITION = -2;

  /** 状态需要向后传播 */
  static final int PROPAGATE = -3;

  volatile int waitStatus;

  volatile Node prev;

  volatile Node next;

  volatile Thread thread;

  Node nextWaiter;
}

//独占模式下获取资源
public final void acquire(int arg) {
  if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    selfInterrupt();
}

private Node addWaiter(Node mode) {
  Node node = new Node(Thread.currentThread(), mode);
  Node pred = tail;
  //尝试快速入队
  if (pred != null) { //队列已经初始化
    node.prev = pred;
    if (compareAndSetTail(pred, node)) {
      pred.next = node;
      return node; //快速入队成功后，就直接返回了
    }
  }
  //快速入队失败，也就是说队列都还每初始化
  enq(node);
  return node;
}

//执行入队
private Node enq(final Node node) {
  for (;;) {
    Node t = tail;
    // 如果没有队尾即还没初始化
    if (t == null) { 
      //如果队列为空，用一个空节点充当队列头
      if (compareAndSetHead(new Node()))
        //尾部指针也指向队列头
        tail = head;
    } else {
      //队列已经初始化，入队
      node.prev = t;
      if (compareAndSetTail(t, node)) {
        t.next = node;
        //打断循环
        return t;
      }
    }
  }
}

final boolean acquireQueued(final Node node, int arg) {
  boolean failed = true;
  try {
    boolean interrupted = false;
    for (;;) {
      //拿到node的上一个节点
      final Node p = node.predecessor();
      //前置节点为head，说明可以尝试获取资源。排队成功后，尝试拿锁
      if (p == head && tryAcquire(arg)) {
        //获取成功，更新head节点
        setHead(node);
        p.next = null; // help GC
        failed = false;
        return interrupted;
      }
      //尝试拿锁失败后，根据条件进行park
      if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
        interrupted = true;
    }
  } finally {
    if (failed)
      cancelAcquire(node);
  }
}
//获取资源失败后，检测并更新等待状态
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  int ws = pred.waitStatus;
  if (ws == Node.SIGNAL)
    return true;
  if (ws > 0) {
    do {
      //如果前节点取消了，那就往前找到一个等待状态的接替你，并排在它的后面
      node.prev = pred = pred.prev;
    } while (pred.waitStatus > 0);
    pred.next = node;
  } else {
    compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  }
  return false;
}
//阻塞当前线程，返回中断状态
private final boolean parkAndCheckInterrupt() {
  LockSupport.park(this);
  return Thread.interrupted();
}

// 公平锁的实现
protected final boolean tryAcquire(int acquires) {
  final Thread current = Thread.currentThread();
  int c = getState();
  if (c == 0) {
    if (!hasQueuedPredecessors() &&
        compareAndSetState(0, acquires)) {
      setExclusiveOwnerThread(current);
      return true;
    }
  }
  else if (current == getExclusiveOwnerThread()) {
    int nextc = c + acquires;
    if (nextc < 0)
      throw new Error("Maximum lock count exceeded");
    setState(nextc);
    return true;
  }
  return false;
}

// 非公平锁的实现
final boolean nonfairTryAcquire(int acquires) {
  final Thread current = Thread.currentThread();
  int c = getState();
  if (c == 0) {
    if (compareAndSetState(0, acquires)) {
      setExclusiveOwnerThread(current);
      return true;
    }
  }
  else if (current == getExclusiveOwnerThread()) {
    int nextc = c + acquires;
    if (nextc < 0) // overflow
      throw new Error("Maximum lock count exceeded");
    setState(nextc);
    return true;
  }
  return false;
}

while(state 状态不允许获取) {
  if(队列中还没有此线程) {
    入队并阻塞
  }
}
当前线程出队

if(state 状态允许了) {
  恢复阻塞的线程(s) 
}

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();
}

public class ThreadLocalDemo implements Runnable {
  private static ThreadLocal<Integer> threadLocal = new ThreadLocal<Integer>() {
    @Override
    protected Integer initialValue() {
      return 0;
    }
  };


  @Override
  public void run() {
    threadLocal.set(threadLocal.get()+1);
  }
}

// 计算多行数字合
public class CountDownLatchDemo {

  private  CountDownLatch countDownLatch;

  private int storeNum[];

  public CountDownLatchDemo(final int numsLines) {
    this.countDownLatch = new CountDownLatch(numsLines);
    this.storeNum = new int[numsLines];
  }

  //计算每行数字合
  private void calc(int index, int calcNums[]){
    storeNum[index] = 0;
    for (int i = 0; i < calcNums.length; i++) {
      storeNum[index] += calcNums[i];
    }
    countDownLatch.countDown();
  }

  //总计
  private int calcSum() throws InterruptedException {
    countDownLatch.await();
    int result = 0;

    for (int i = 0; i < storeNum.length; i++) {
      result+= storeNum[i];
    }
    return result;
  }

  public static void main(String[] args) throws InterruptedException {

    CountDownLatchDemo countDownLatchDemo = new CountDownLatchDemo(nums.length);
    for (int i = 0; i < nums.length; i++) {
      int finalI = i;
      new Thread(new Runnable() {
        @Override
        public void run() {
          countDownLatchDemo.calc(finalI, nums[finalI]);
        }
      }).start();
    }
    int sum = countDownLatchDemo.calcSum();
    System.out.println(sum);
  }
}

public CountDownLatch(int count) {
  if (count < 0) throw new IllegalArgumentException("count < 0");
  this.sync = new Sync(count);
}

Sync(int count) {
  setState(count);
}

public void await() throws InterruptedException {
  sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
  if (Thread.interrupted())
    throw new InterruptedException();
  // tryAcquireShared(arg) ==》return (getState() == 0) ? 1 : -1;
  // 即当state不为0时进入，为0直接运行不阻塞
  if (tryAcquireShared(arg) < 0)
    doAcquireSharedInterruptibly(arg);
}

private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
  // 加入一个共享队列元素
  final Node node = addWaiter(Node.SHARED);
  boolean failed = true;
  try {
    for (;;) {
      final Node p = node.predecessor();
      if (p == head) {
        // tryAcquireShared(arg) ==》(getState() == 0) ? 1 : -1
        // 即如果是队列首元素就等待释放
        int r = tryAcquireShared(arg);
        if (r >= 0) {
          setHeadAndPropagate(node, r);
          p.next = null; // help GC
          failed = false;
          return;
        }
      }
      if (shouldParkAfterFailedAcquire(p, node) &&
          parkAndCheckInterrupt())
        throw new InterruptedException();
    }
  } finally {
    if (failed)
      cancelAcquire(node);
  }
}
}

public class CyclicBarrierDemo {

  public void meet(CyclicBarrier cyclicBarrier){
    String threadName = Thread.currentThread().getName();
    System.out.println(threadName + " 到场 ");
    try {
      Thread.sleep(1000);
      cyclicBarrier.await();
    } catch (InterruptedException e) {
      // 当前线程被中断
      e.printStackTrace();
    } catch (BrokenBarrierException e) {
      // 不是被中断的线程，但是其它相关线程await()时发生中断
      e.printStackTrace();
    }
    System.out.println(threadName + " 准备开会发言");
  }

  public static void main(String[] args) {
    CyclicBarrierDemo cyclicBarrierDemo = new CyclicBarrierDemo();
    CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
    for (int i = 0; i < 5; i++) {
      new Thread(new Runnable() {
        @Override
        public void run() {
          cyclicBarrierDemo.meet(cyclicBarrier);
        }
      }).start();
    }
  }
}

public class SemaphoreDemo {

  private Semaphore semaphore = new Semaphore(5);

  public void doSomething() {
    try {
      /**
        * 在 semaphore.acquire() 和 semaphore.release()之间的代码，同一时刻只允许制定个数的线程进入，
        * 因为semaphore的构造方法是5，则同一时刻只允许5个线程进入，其他线程只能等待。
        **/
      semaphore.acquire();
      System.out.println(Thread.currentThread().getName() + ":doSomething start-" + new Date());
      Thread.sleep(2000);
      System.out.println(Thread.currentThread().getName() + ":doSomething end-" + new Date());
      semaphore.release();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  public static void main(String[] args) {
    SemaphoreDemo service = new SemaphoreDemo();
    for (int i = 0; i < 20; i++) {
      new Thread(new Runnable() {
        @Override
        public void run() {
          service.doSomething();
        }
      }).start();
    }
  }
}

// 每次计算完成交换结果
public class ExchangerTest {
  static class Producer extends Thread {
    private Exchanger<Integer> exchanger;
    private static int data = 0;
    Producer(Exchanger<Integer> exchanger) {
      super("Producer");
      this.exchanger = exchanger;
    }

    @Override
    public void run() {
      for (int i=1; i<5; i++) {
        try {
          TimeUnit.SECONDS.sleep(1);
          data = i;
          System.out.println(getName()+" 交换前:" + data);
          data = exchanger.exchange(data);
          System.out.println(getName()+" 交换后:" + data);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }

  static class Consumer extends Thread {
    private Exchanger<Integer> exchanger;
    private static int data = 0;
    Consumer(Exchanger<Integer> exchanger) {
      super("Consumer");
      this.exchanger = exchanger;
    }

    @Override
    public void run() {
      while (true) {
        data = 0;
        System.out.println(getName()+" 交换前:" + data);
        try {
          TimeUnit.SECONDS.sleep(1);
          data = exchanger.exchange(data);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        System.out.println(getName()+" 交换后:" + data);
      }
    }
  }

  public static void main(String[] args) throws InterruptedException {
    Exchanger<Integer> exchanger = new Exchanger<Integer>();
    new Producer(exchanger).start();
    new Consumer(exchanger).start();
  }
}

public class FutureDemo {

  private static class Product implements Callable<Product> {
    @Override
    public Product call() throws Exception {
      System.out.println("生成产品中...");
      Thread.sleep(5000);
      System.out.println("生成产品结束...");
      return new Product();
    }
  }

  public static void main(String[] args) {
    FutureTask<Product> futureTask = new FutureTask<>(new Product());
    // 使用FutureTask创建一个thread，FutureTask实现了Runnable接口
    Thread thread = new Thread(futureTask);
    // 启动线程任务
    thread.start();
    // 此时FutureTask任务在其它线程中执行，主线程不受影响
    System.out.println("干点其他事...");
    try {
      // 此时需要FutureTask任务执行的结果，会在此一直等候
      Product product = futureTask.get();
    } catch (InterruptedException e) {
      System.out.println("发生中断");
    } catch (ExecutionException e) {
      System.out.println("执行过程中发生异常 " + e.getMessage());
    }
  }
}

public boolean add(E e) {
  final ReentrantLock lock = this.lock;
  lock.lock();
  try {
    Object[] elements = getArray();
    int len = elements.length;
    Object[] newElements = Arrays.copyOf(elements, len + 1);
    newElements[len] = e;
    setArray(newElements);
    return true;
  } finally {
    lock.unlock();
  }
}

/**
 * corePoolSize:线程池的核心大小，当提交一个任务到线程池时，线程池会创建一个线程来执行任务
 *              即使其他空闲的基本线程能够执行新任务也会创建线程，如需要执行的任务数大于线
 *              程池基本大小时就不再创建。如果调用了线程池的prestartAllCoreThreads()方法
 *              线程池会提前创建并启动所有基本线程。
 * maximumPoolSize:线程池最大大小，线程池允许创建的最大线程数。如果队列满了，并且已创建的线程数
 *                 小于最大线程数大于等于核心线程数，则线程池会再创建新的线程执行任务。如果使用
 *                 了无界的任务队列这个参数就没什么效果。
 * keepAliveTime:救急线程活动保持时间，线程池的工作线程空闲后，保持存活的时间。所以如果任务很多
 *               并且每个任务执行的时间比较短，可以调大这个时间，提高线程的利用率。
 * TimeUnit:救急线程活动保持时间的单位，可选的单位有天(DAYS)，小时(HOURS)，分钟(MINUTES)，
 *          毫秒(MILLISECONDS)，微秒(MICROSECONDS, 千分之一毫秒)和毫微秒(NANOSECONDS, 
 *          千分之一微秒)
 * workQueue:任务对列，用于保存等待执行的任务的阻塞队列
 *           - ArrayBlockingQueue：基于数组结构的有界阻塞队列
 *           - LinkedBlockingQueue：基于链表的阻塞队列,如果没构造函数没传入队列大小则为无界队列
 *                                  Executors.newFixedThreadPool()使用了这个队列
 *           - SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用
 *                               移除操作，否则插入操作一直处于阻塞状态
 *                               Executors.newCachedThreadPool使用了这个队列
 *           - PriorityBlockingQueue：个具有优先级得无限阻塞队列
 * threadFactory:用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字，
 *               Debug和定位问题时非常有帮助。
 * handler:当队列和线程池都满了，必须采取一种策略处理提交的新任务。
 *         - AbortPolicy：默认策略，无法处理新任务时抛出异常
 *         - CallerRunsPolicy：使用调用者所在线程来运行任务
 *         - DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务
 *         - DiscardPolicy：不处理，丢弃掉
 */
public ThreadPoolExecutor(int corePoolSize, 
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler);

public class ThreadPoolExecutor extends AbstractExecutorService {

  public void execute(Runnable command) {
    if (command == null)
      throw new NullPointerException();
    // 获取线程池控制状态
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) { // worker数量小于corePoolSize
      if (addWorker(command, true)) // 添加worker
        return;
      // 不成功则再次获取线程池控制状态
      c = ctl.get();
    }
    // 线程池处于RUNNING状态，将用户自定义的Runnable对象添加进workQueue队列
    if (isRunning(c) && workQueue.offer(command)) { 
      // 再次检查，获取线程池控制状态
      int recheck = ctl.get();
      // 线程池不处于RUNNING状态，将自定义任务从workQueue队列中移除
      if (! isRunning(recheck) && remove(command)) 
        // 拒绝执行命令
        reject(command);
      else if (workerCountOf(recheck) == 0) // worker数量等于0
        // 添加worker
        addWorker(null, false);
    }
    else if (!addWorker(command, false)) // 添加worker失败
      // 拒绝执行命令
      reject(command);
  }
}
private boolean addWorker(Runnable firstTask, boolean core) {
  //...
  if (workerAdded) { // worker被添加
    // 开始执行worker的run方法，调用线程start启动线程
    t.start();  
    // 设置worker已开始标识
    workerStarted = true;
  }
  //...
  return workerStarted;
}

private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
  public void run() {
    runWorker(this);
  } 
  final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
      while (task != null || (task = getTask()) != null) {
        w.lock();
        if ((runStateAtLeast(ctl.get(), STOP) ||
             (Thread.interrupted() &&
              runStateAtLeast(ctl.get(), STOP))) &&
            !wt.isInterrupted())
          wt.interrupt();
        try {
          beforeExecute(wt, task);
          Throwable thrown = null;
          try {
            // 执行run方法
            task.run();
          } catch (RuntimeException x) {
            thrown = x; throw x;
          } catch (Error x) {
            thrown = x; throw x;
          } catch (Throwable x) {
            thrown = x; throw new Error(x);
          } finally {
            afterExecute(task, thrown);
          }
        } finally {
          task = null;
          w.completedTasks++;
          w.unlock();
        }
      }
      completedAbruptly = false;
    } finally {
      processWorkerExit(w, completedAbruptly);
    }
  }
}

public class Executors {

  // 创建一个立即执行无缓冲的线程池
  public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());
  }
  
  // 创建一个固定大小的线程池，corePoolSize和maximumPoolSize相等即不会在队列满了创建线程
  // LinkedBlockingQueue默认构造器为无界队列，队列不可能满
  public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
  }
  
  // 创建一个单一线程运行的线程池，corePoolSize和maximumPoolSize相等即不会在队列满了创建线程
  // LinkedBlockingQueue默认构造器为无界队列，队列不可能满
  public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,
                 0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));
  }
  
  // 创建一个可以线程可以定时调度的线程池
  public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
  }
  
  public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements 
        ScheduledExecutorService
    public ScheduledThreadPoolExecutor(int corePoolSize) {
      super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, 
              new DelayedWorkQueue());
    }
  }

  // 创建一个可执行Fork/Join任务的线程池
  public static ExecutorService newWorkStealingPool(int parallelism) {
    return new ForkJoinPool(parallelism,
       ForkJoinPool.defaultForkJoinWorkerThreadFactory,null, true);
  }
}

/*
  线程池状态变为 SHUTDOWN
   - 不会接收新任务
   - 但已提交任务会执行完
   - 此方法不会阻塞调用线程的执行
*/
public void shutdown() {
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    checkShutdownAccess();
    // 修改线程池状态
    advanceRunState(SHUTDOWN);
    // 仅会打断空闲线程
    interruptIdleWorkers();
    onShutdown(); // 扩展点 ScheduledThreadPoolExecutor
  } finally {
    mainLock.unlock();
  }
  // 尝试终结(没有运行的线程可以立刻终结，如果还有运行的线程也不会等)
  tryTerminate();
}

/*
  线程池状态变为 STOP
   - 不会接收新任务
   - 会将队列中的任务返回
   - 并用 interrupt 的方式中断正在执行的任务
*/
public List<Runnable> shutdownNow() {
  List<Runnable> tasks;
  final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
  try {
    checkShutdownAccess();
    // 修改线程池状态
    advanceRunState(STOP);
    // 打断所有线程
    interruptWorkers();
    // 获取队列中剩余任务
    tasks = drainQueue();
  } finally {
    mainLock.unlock();
  }
  // 尝试终结
  tryTerminate();
  return tasks;
}

// 本地方法，返回线程中断状态，并且根据参数判断是否需要清除标识位
private native boolean isInterrupted(boolean ClearInterrupted);

// 返回线程中断状态，不清空标识
public boolean isInterrupted() {return isInterrupted(false);}

// 静态方法，返回当前线程中断状态，清空标识
public static boolean interrupted() { return currentThread().isInterrupted(true);}

public class InterruptTest{

  private static ReentrantLock reentrantLock = new ReentrantLock();

  private static class InterruptRunnable implements Runnable {
    @Override
    public void run() {
      String currentThreadName =  Thread.currentThread().getName();
      reentrantLock.lock();
      while (!Thread.interrupted()) {
        System.out.println(currentThreadName + " do business logic");
        for (int i = 0; i < 2000000000; i++) {

        }
        System.out.println(currentThreadName + " do business logic over");
      }
      // 线程发生中断
      System.out.println(currentThreadName + " happen interrupt do something");
      // 必须释放资源否则“T-2”线程一直阻塞
      reentrantLock.unlock();
      // 释放资源
      System.out.println(currentThreadName + " release resources");
    }
  }

  public static void main(String[] args) throws InterruptedException {
    Thread thread = new Thread(new InterruptRunnable(),"T-1");
    thread.start();
    TimeUnit.SECONDS.sleep(1);
    Thread thread2 = new Thread(new InterruptRunnable(), "T-2");
    thread2.start();
    thread.interrupt();
    // 如果调用的是thread.stop()可以释放synchronized资源但是不会释放Lock实现类的资源
  }
}

/**
 * 步骤1：为new出来的对象开辟内存空间
 * 步骤2：初始化，执行构造器方法的逻辑代码片段
 * 步骤3：完成obj引用的赋值操作，将其指向刚刚开辟的内存地址
 * 这三个步骤可能进行指令重排序变为：步骤1、步骤3、步骤2
 * 如果发生重排序可能obj不为null但是未初始化完成
 */
ObjectVarilabe obj = new ObjectVarilabe()

public class Singleton {

  private static volatile Singleton instance;

  private Singleton() {
  }

  public static Singleton getInstance() {
    if(instance==null){                    
      synchronized (Singleton.class){    
        if(instance==null){           
          instance=new Singleton();
        }
      }
    }
    return instance;
  }
}

public class Demo {

  private int balance;

  private StampedLock lock = new StampedLock();

  //===================================读写锁转化====================================
  public void conditionReadWrite (int value) {
    // 首先加独占读锁判断balance的值是否符合更新的条件
    long stamp = lock.readLock();
    try{
      while (balance > 0) {
        // 下面需要进行写操作，所以转化为写锁
        // ReentrantReadWriteLock是直接加写锁，锁升级
        long writeStamp = lock.tryConvertToWriteLock(stamp);
        if(writeStamp != 0) { // 成功转换成为写锁
          stamp = writeStamp;
          balance += value;
          break;
        } else {
          // 没有转换成写锁，这里需要首先释放读锁，然后再拿到写锁
          lock.unlockRead(stamp);

          //。。。

          // 获取写锁
          stamp = lock.writeLock();
        }
      }
    }finally{
      lock.unlock(stamp); 
    }
  }

  //====================================乐观读操作===================================
  public void optimisticRead() {
    // 获取乐观读锁
    long stamp = lock.tryOptimisticRead();
    int c = balance;
    // 这里其它可能会出现了写操作，因此要进行判断读锁获取的是否有问题
    if(!lock.validate(stamp)) {
      // 要从新读取
      long readStamp = lock.readLock();
      c = balance;
      stamp = readStamp;
    }
    lock.unlockRead(stamp);
  }

  //==============以下read/write操作和ReentrantReadWriteLock效果完全一样==============
  public void read () {
    long stamp = lock.readLock();
    lock.tryOptimisticRead();
    int c = balance;
    // ...
    lock.unlockRead(stamp);
  }

  public void write(int value) {
    long stamp = lock.writeLock();
    balance += value;
    lock.unlockWrite(stamp);
  }
}