多线程是实现并发机制的一种有效手段。在 Java 中实现多线程有两种手段，一种是继承 Thread 类，另一种就是实现 Runnable/Callable 接口。
java.util.concurrent 包是专为 Java并发编程而设计的包。类图如下：
一、同步
1.1 synchronized 关键字，用来给对象和方法或者代码块加锁。
1同步方法 2synchronized T methodName(){} 3同步方法锁定的是当前对象。当多线程通过同一个对象引用多次调用当前同步方法时，需同步执行。静态同步方法，锁的是当前类型的类对象。同步方法只影响锁定同一个锁对象的同步方法。不影响其他线程调用非同步方法，或调用其他锁资源的同步方法。锁可重入。 同一个线程，多次调用同步代码，锁定同一个锁对象，可重入。子类同步方法覆盖父类同步方法。可以指定调用父类的同步方法，相当于锁的重入。当同步方法中发生异常的时候，自动释放锁资源。不会影响其他线程的执行。 4同步代码块 5T methodName(){ 6 synchronized(object){} 7} 8同步代码块在执行时，是锁定 object 对象。当多个线程调用同一个方法时，锁定对象不变的情况下，需同步执行。 9同步代码块的同步粒度更加细致，效率更高。 10 11T methodName(){ 12 synchronized(this){} 13} 14当锁定对象为 this 时，相当于同步方法。
同步代码一旦加锁后，那么会有一个临时的锁引用执行锁对象，和真实的引用无直接关联。在锁未释放之前，修改锁对象引用，不会影响同步代码的执行。
Java 虚拟机中的同步(Synchronization)基于进入和退出管程(Monitor)对象实现。同步方法 并不是由 monitor enter 和 monitor exit 指令来实现同步的，而是由方法调用指令读取运行时常量池中方法的 ACC_SYNCHRONIZED 标志来隐式实现的。在 Java 虚拟机(HotSpot)中，monitor 是由 ObjectMonitor 实现的。
1 1 /** 2 2 * synchronized关键字 3 3 * 锁对象。synchronized(this)和synchronized方法都是锁当前对象。 4 4 */ 5 5 package concurrent.t01; 6 6 7 7 import java.util.concurrent.TimeUnit; 8 8 9 9 public class Test_01 { 1010 private int count = 0; 1111 private Object o = new Object(); 1212 1313 public void testSync1(){ 1414 synchronized(o){ 1515 System.out.println(Thread.currentThread().getName() 1616 + " count = " + count++); 1717 } 1818 } 1919 2020 public void testSync2(){ 2121 synchronized(this){ 2222 System.out.println(Thread.currentThread().getName() 2323 + " count = " + count++); 2424 } 2525 } 2626 2727 public synchronized void testSync3(){ 2828 System.out.println(Thread.currentThread().getName() 2929 + " count = " + count++); 3030 try { 3131 TimeUnit.SECONDS.sleep(3); 3232 } catch (InterruptedException e) { 3333 e.printStackTrace(); 3434 } 3535 } 3636 3737 public static void main(String[] args) { 3838 final Test_01 t = new Test_01(); 3939 new Thread(new Runnable() { 4040 @Override 4141 public void run() { 4242 t.testSync3(); 4343 } 4444 }).start(); 4545 new Thread(new Runnable() { 4646 @Override 4747 public void run() { 4848 t.testSync3(); 4949 } 5050 }).start(); 5151 } 5252 5353 }
synchronized关键字
1.2 volatile 关键字
变量的线程可见性。在 CPU 计算过程中，会将计算过程需要的数据加载到 CPU 计算缓存中，当 CPU 计算中断时，有可能刷新缓存，重新读取内存中的数据。在线程运行的过程中，如果某变量被其他线程修改，可能造成数据不一致的情况，从而导致结果错误。而 volatile修饰的变量是线程可见的，当 JVM 解释 volatile 修饰的变量时，会通知 CPU，在计算过程中，每次使用变量参与计算时，都会检查内存中的数据是否发生变化，而不是一直使用 CPU 缓存中的数据，可以保证计算结果的正确。volatile 只是通知底层计算时，CPU 检查内存数据，而不是让一个变量在多个线程中同步。
1 1 /** 2 2 * volatile关键字 3 3 * volatile的可见性 4 4 * 通知OS操作系统底层，在CPU计算过程中，都要检查内存中数据的有效性。保证最新的内存数据被使用。 5 5 * 6 6 */ 7 7 package concurrent.t01; 8 8 9 9 import java.util.concurrent.TimeUnit; 1010 1111 public class Test_09 { 1212 1313 volatile boolean b = true; 1414 1515 void m(){ 1616 System.out.println("start"); 1717 while(b){} 1818 System.out.println("end"); 1919 } 2020 2121 public static void main(String[] args) { 2222 final Test_09 t = new Test_09(); 2323 new Thread(new Runnable() { 2424 @Override 2525 public void run() { 2626 t.m(); 2727 } 2828 }).start(); 2929 3030 try { 3131 TimeUnit.SECONDS.sleep(1); 3232 } catch (InterruptedException e) { 3333 // TODO Auto-generated catch block 3434 e.printStackTrace(); 3535 } 3636 3737 t.b = false; 3838 } 3939 4040 }
volatile的可见性
1 1 /** 2 2 * volatile关键字 3 3 * volatile的非原子性问题 4 4 * volatile， 只能保证可见性，不能保证原子性。 5 5 * 不是枷锁问题，只是内存数据可见。 6 6 */ 7 7 package concurrent.t01; 8 8 9 9 import java.util.ArrayList; 1010 import java.util.List; 1111 1212 public class Test_10 { 1313 1414 volatile int count = 0; 1515 /*synchronized*/ void m(){ 1616 for(int i = 0; i < 10000; i++){ 1717 count++; 1818 } 1919 } 2020 2121 public static void main(String[] args) { 2222 final Test_10 t = new Test_10(); 2323 List<Thread> threads = new ArrayList<>(); 2424 for(int i = 0; i < 10; i++){ 2525 threads.add(new Thread(new Runnable() { 2626 @Override 2727 public void run() { 2828 t.m(); 2929 } 3030 })); 3131 } 3232 for(Thread thread : threads){ 3333 thread.start(); 3434 } 3535 for(Thread thread : threads){ 3636 try { 3737 thread.join(); 3838 } catch (InterruptedException e) { 3939 // TODO Auto-generated catch block 4040 e.printStackTrace(); 4141 } 4242 } 4343 System.out.println(t.count); 4444 } 4545 }
volatile的非原子性问题
1.3 wait¬ify
当线程执行wait()方法时候，会释放当前的锁，然后让出CPU，进入等待状态。只有当 notify/notifyAll() 被执行时候，才会唤醒一个或多个正处于等待状态的线程，然后继续往下执行，直到执行完synchronized 代码块的代码或是中途遇到wait() ，再次释放锁。
由于 wait()、notify/notifyAll() 在synchronized 代码块执行，说明当前线程一定是获取了锁的。wait()、notify/notifyAll() 方法是Object的本地final方法，无法被重写。
1 1 /** 2 2 * 生产者消费者 3 3 * wait¬ify 4 4 * wait/notify都是和while配合应用的。可以避免多线程并发判断逻辑失效问题。各位想想为什么不能用if 5 5 */ 6 6 package concurrent.t04; 7 7 8 8 import java.util.LinkedList; 9 9 import java.util.concurrent.TimeUnit; 1010 1111 public class TestContainer01<E> { 1212 1313 private final LinkedList<E> list = new LinkedList<>(); 1414 private final int MAX = 10; 1515 private int count = 0; 1616 1717 public synchronized int getCount(){ 1818 return count; 1919 } 2020 2121 public synchronized void put(E e){ 2222 while(list.size() == MAX){ 2323 try { 2424 this.wait(); 2525 } catch (InterruptedException e1) { 2626 e1.printStackTrace(); 2727 } 2828 } 2929 3030 list.add(e); 3131 count++; 3232 this.notifyAll(); 3333 } 3434 3535 public synchronized E get(){ 3636 E e = null; 3737 while(list.size() == 0){ 3838 try{ 3939 this.wait(); 4040 } catch (InterruptedException e1) { 4141 e1.printStackTrace(); 4242 } 4343 } 4444 e = list.removeFirst(); 4545 count--; 4646 this.notifyAll(); 4747 return e; 4848 } 4949 5050 public static void main(String[] args) { 5151 final TestContainer01<String> c = new TestContainer01<>(); 5252 for(int i = 0; i < 10; i++){ 5353 new Thread(new Runnable() { 5454 @Override 5555 public void run() { 5656 for(int j = 0; j < 5; j++){ 5757 System.out.println(c.get()); 5858 } 5959 } 6060 }, "consumer"+i).start(); 6161 } 6262 try { 6363 TimeUnit.SECONDS.sleep(2); 6464 } catch (InterruptedException e) { 6565 // TODO Auto-generated catch block 6666 e.printStackTrace(); 6767 } 6868 for(int i = 0; i < 2; i++){ 6969 new Thread(new Runnable() { 7070 @Override 7171 public void run() { 7272 for(int j = 0; j < 25; j++){ 7373 c.put("container value " + j); 7474 } 7575 } 7676 }, "producer"+i).start(); 7777 } 7878 } 7979 8080 }
wait¬ify
1.4 AtomicXxx 类型
原子类型。
在 concurrent.atomic 包中定义了若干原子类型，这些类型中的每个方法都是保证了原子操作的。多线程并发访问原子类型对象中的方法，不会出现数据错误。在多线程开发中，如果某数据需要多个线程同时操作，且要求计算原子性，可以考虑使用原子类型对象。
- AtomicBoolean，AtomicInteger，AtomicLong，AtomicReference
- AtomicIntegerArray，AtomicLongArray
- AtomicLongFieldUpdater，AtomicIntegerFieldUpdater，AtomicReferenceFieldUpdater
- AtomicMarkableReference，AtomicStampedReference，AtomicReferenceArray
注意：原子类型中的方法 是保证了原子操作，但多个方法之间是没有原子性的。即访问对2个或2个以上的atomic变量（或者对单个atomic变量进行2次或2次以上的操作），还是需要同步。
1 1 /** 2 2 * AtomicXxx 3 3 * 同步类型 4 4 * 原子操作类型。 其中的每个方法都是原子操作。可以保证线程安全。 5 5 */ 6 6 package concurrent.t01; 7 7 8 8 import java.util.ArrayList; 9 9 import java.util.List; 1010 import java.util.concurrent.atomic.AtomicInteger; 1111 1212 public class Test_11 { 1313 AtomicInteger count = new AtomicInteger(0); 1414 void m(){ 1515 for(int i = 0; i < 10000; i++){ 1616 /*if(count.get() < 1000)*/ 1717 count.incrementAndGet(); 1818 } 1919 } 2020 2121 public static void main(String[] args) { 2222 final Test_11 t = new Test_11(); 2323 List<Thread> threads = new ArrayList<>(); 2424 for(int i = 0; i < 10; i++){ 2525 threads.add(new Thread(new Runnable() { 2626 @Override 2727 public void run() { 2828 t.m(); 2929 } 3030 })); 3131 } 3232 for(Thread thread : threads){ 3333 thread.start(); 3434 } 3535 for(Thread thread : threads){ 3636 try { 3737 thread.join(); 3838 } catch (InterruptedException e) { 3939 // TODO Auto-generated catch block 4040 e.printStackTrace(); 4141 } 4242 } 4343 System.out.println(t.count.intValue()); 4444 } 4545 }
AtomicXxx
1.5 CountDownLatch 门闩
门闩是 concurrent 包中定义的一个类型，是用于多线程通讯的一个辅助类型。门闩相当于在一个门上加多个锁，当线程调用 await 方法时，会检查门闩数量，如果门闩数量大于 0，线程会阻塞等待。当线程调用 countDown 时，会递减门闩的数量，当门闩数量为 0 时，await 阻塞线程可执行。
1 1 /** 2 2 * 门闩 - CountDownLatch 3 3 * 可以和锁混合使用，或替代锁的功能。 4 4 * 在门闩未完全开放之前等待。当门闩完全开放后执行。 5 5 * 避免锁的效率低下问题。 6 6 */ 7 7 package concurrent.t01; 8 8 9 9 import java.util.concurrent.CountDownLatch; 1010 import java.util.concurrent.TimeUnit; 1111 1212 public class Test_15 { 1313 CountDownLatch latch = new CountDownLatch(5); 1414 1515 void m1(){ 1616 try { 1717 latch.await();// 等待门闩开放。 1818 } catch (InterruptedException e) { 1919 e.printStackTrace(); 2020 } 2121 System.out.println("m1() method"); 2222 } 2323 2424 void m2(){ 2525 for(int i = 0; i < 10; i++){ 2626 if(latch.getCount() != 0){ 2727 System.out.println("latch count : " + latch.getCount()); 2828 latch.countDown(); // 减门闩上的锁。 2929 } 3030 try { 3131 TimeUnit.MILLISECONDS.sleep(500); 3232 } catch (InterruptedException e) { 3333 // TODO Auto-generated catch block 3434 e.printStackTrace(); 3535 } 3636 System.out.println("m2() method : " + i); 3737 } 3838 } 3939 4040 public static void main(String[] args) { 4141 final Test_15 t = new Test_15(); 4242 new Thread(new Runnable() { 4343 @Override 4444 public void run() { 4545 t.m1(); 4646 } 4747 }).start(); 4848 4949 new Thread(new Runnable() { 5050 @Override 5151 public void run() { 5252 t.m2(); 5353 } 5454 }).start(); 5555 } 5656 5757 }
CountDownLatch
1.6 锁的重入
在 Java 中，同步锁是可以重入的。只有同一线程调用同步方法或执行同步代码块，对同一个对象加锁时才可重入。
当线程持有锁时，会在 monitor 的计数器中执行递增计算，若当前线程调用其他同步代码，且同步代码的锁对象相同时，monitor 中的计数器继续递增。每个同步代码执行结束，monitor 中的计数器都会递减，直至所有同步代码执行结束，monitor 中的计数器为 0 时，释放锁标记，_Owner 标记赋值为 null。
1 1 /** 2 2 * 锁可重入。 同一个线程，多次调用同步代码，锁定同一个锁对象，可重入。 3 3 */ 4 4 package concurrent.t01; 5 5 6 6 import java.util.concurrent.TimeUnit; 7 7 8 8 public class Test_06 { 9 9 1010 synchronized void m1(){ // 锁this 1111 System.out.println("m1 start"); 1212 try { 1313 TimeUnit.SECONDS.sleep(2); 1414 } catch (InterruptedException e) { 1515 e.printStackTrace(); 1616 } 1717 m2(); 1818 System.out.println("m1 end"); 1919 } 2020 synchronized void m2(){ // 锁this 2121 System.out.println("m2 start"); 2222 try { 2323 TimeUnit.SECONDS.sleep(1); 2424 } catch (InterruptedException e) { 2525 e.printStackTrace(); 2626 } 2727 System.out.println("m2 end"); 2828 } 2929 3030 public static void main(String[] args) { 3131 3232 new Test_06().m1(); 3333 3434 } 3535 3636 }
锁可重入
1.7 ReentrantLock
重入锁，建议应用的同步方式。相对效率比 synchronized 高。量级较轻。使用重入锁， 必须手工释放锁标记。一般都是在 finally 代码块中定义释放锁标记的 unlock 方法。
1 1 /** 2 2 * ReentrantLock 3 3 * 重入锁 4 4 */ 5 5 package concurrent.t03; 6 6 7 7 import java.util.concurrent.TimeUnit; 8 8 import java.util.concurrent.locks.Lock; 9 9 import java.util.concurrent.locks.ReentrantLock; 1010 1111 public class Test_01 { 1212 Lock lock = new ReentrantLock(); 1313 1414 void m1(){ 1515 try{ 1616 lock.lock(); // 加锁 1717 for(int i = 0; i < 10; i++){ 1818 TimeUnit.SECONDS.sleep(1); 1919 System.out.println("m1() method " + i); 2020 } 2121 }catch(InterruptedException e){ 2222 e.printStackTrace(); 2323 }finally{ 2424 lock.unlock(); // 解锁 2525 } 2626 } 2727 2828 void m2(){ 2929 lock.lock(); 3030 System.out.println("m2() method"); 3131 lock.unlock(); 3232 } 3333 3434 public static void main(String[] args) { 3535 final Test_01 t = new Test_01(); 3636 new Thread(new Runnable() { 3737 @Override 3838 public void run() { 3939 t.m1(); 4040 } 4141 }).start(); 4242 try { 4343 TimeUnit.SECONDS.sleep(1); 4444 } catch (InterruptedException e) { 4545 e.printStackTrace(); 4646 } 4747 new Thread(new Runnable() { 4848 @Override 4949 public void run() { 5050 t.m2(); 5151 } 5252 }).start(); 5353 } 5454 }
重入锁
1 1 /** 2 2 * 尝试锁 3 3 */ 4 4 package concurrent.t03; 5 5 6 6 import java.util.concurrent.TimeUnit; 7 7 import java.util.concurrent.locks.Lock; 8 8 import java.util.concurrent.locks.ReentrantLock; 9 9 1010 public class Test_02 { 1111 Lock lock = new ReentrantLock(); 1212 1313 void m1(){ 1414 try{ 1515 lock.lock(); 1616 for(int i = 0; i < 10; i++){ 1717 TimeUnit.SECONDS.sleep(1); 1818 System.out.println("m1() method " + i); 1919 } 2020 }catch(InterruptedException e){ 2121 e.printStackTrace(); 2222 }finally{ 2323 lock.unlock(); 2424 } 2525 } 2626 2727 void m2(){ 2828 boolean isLocked = false; 2929 try{ 3030 // 尝试锁， 如果有锁，无法获取锁标记，返回false。 3131 // 如果获取锁标记，返回true 3232 // isLocked = lock.tryLock(); 3333 3434 // 阻塞尝试锁，阻塞参数代表的时长，尝试获取锁标记。 3535 // 如果超时，不等待。直接返回。 3636 isLocked = lock.tryLock(5, TimeUnit.SECONDS); 3737 3838 if(isLocked){ 3939 System.out.println("m2() method synchronized"); 4040 }else{ 4141 System.out.println("m2() method unsynchronized"); 4242 } 4343 }catch(Exception e){ 4444 e.printStackTrace(); 4545 }finally{ 4646 if(isLocked){ 4747 // 尝试锁在解除锁标记的时候，一定要判断是否获取到锁标记。 4848 // 如果当前线程没有获取到锁标记，会抛出异常。 4949 lock.unlock(); 5050 } 5151 } 5252 } 5353 5454 public static void main(String[] args) { 5555 final Test_02 t = new Test_02(); 5656 new Thread(new Runnable() { 5757 @Override 5858 public void run() { 5959 t.m1(); 6060 } 6161 }).start(); 6262 try { 6363 TimeUnit.SECONDS.sleep(1); 6464 } catch (InterruptedException e) { 6565 // TODO Auto-generated catch block 6666 e.printStackTrace(); 6767 } 6868 new Thread(new Runnable() { 6969 @Override 7070 public void run() { 7171 t.m2(); 7272 } 7373 }).start(); 7474 } 7575 }
尝试锁
1 1 /** 2 2 * 可打断 3 3 * 4 4 * 阻塞状态： 包括普通阻塞，等待队列，锁池队列。 5 5 * 普通阻塞： sleep(10000)， 可以被打断。调用thread.interrupt()方法，可以打断阻塞状态，抛出异常。 6 6 * 等待队列： wait()方法被调用，也是一种阻塞状态，只能由notify唤醒。无法打断 7 7 * 锁池队列： 无法获取锁标记。不是所有的锁池队列都可被打断。 8 8 * 使用ReentrantLock的lock方法，获取锁标记的时候，如果需要阻塞等待锁标记，无法被打断。 9 9 * 使用ReentrantLock的lockInterruptibly方法，获取锁标记的时候，如果需要阻塞等待，可以被打断。 1010 * 1111 */ 1212 package concurrent.t03; 1313 1414 import java.util.concurrent.TimeUnit; 1515 import java.util.concurrent.locks.Lock; 1616 import java.util.concurrent.locks.ReentrantLock; 1717 1818 public class Test_03 { 1919 Lock lock = new ReentrantLock(); 2020 2121 void m1(){ 2222 try{ 2323 lock.lock(); 2424 for(int i = 0; i < 5; i++){ 2525 TimeUnit.SECONDS.sleep(1); 2626 System.out.println("m1() method " + i); 2727 } 2828 }catch(InterruptedException e){ 2929 e.printStackTrace(); 3030 }finally{ 3131 lock.unlock(); 3232 } 3333 } 3434 3535 void m2(){ 3636 try{ 3737 lock.lockInterruptibly(); // 可尝试打断，阻塞等待锁。可以被其他的线程打断阻塞状态 3838 System.out.println("m2() method"); 3939 }catch(InterruptedException e){ 4040 System.out.println("m2() method interrupted"); 4141 }finally{ 4242 try{ 4343 lock.unlock(); 4444 }catch(Exception e){ 4545 e.printStackTrace(); 4646 } 4747 } 4848 } 4949 5050 public static void main(String[] args) { 5151 final Test_03 t = new Test_03(); 5252 new Thread(new Runnable() { 5353 @Override 5454 public void run() { 5555 t.m1(); 5656 } 5757 }).start(); 5858 try { 5959 TimeUnit.SECONDS.sleep(1); 6060 } catch (InterruptedException e) { 6161 // TODO Auto-generated catch block 6262 e.printStackTrace(); 6363 } 6464 Thread t2 = new Thread(new Runnable() { 6565 @Override 6666 public void run() { 6767 t.m2(); 6868 } 6969 }); 7070 t2.start(); 7171 try { 7272 TimeUnit.SECONDS.sleep(1); 7373 } catch (InterruptedException e) { 7474 // TODO Auto-generated catch block 7575 e.printStackTrace(); 7676 } 7777 t2.interrupt();// 打断线程休眠。非正常结束阻塞状态的线程，都会抛出异常。 7878 } 7979 }
可打断
1 1 /** 2 2 * 公平锁 3 3 */ 4 4 package concurrent.t03; 5 5 6 6 import java.util.concurrent.locks.ReentrantLock; 7 7 8 8 public class Test_04 { 9 9 1010 public static void main(String[] args) { 1111 TestReentrantlock t = new TestReentrantlock(); 1212 //TestSync t = new TestSync(); 1313 Thread t1 = new Thread(t); 1414 Thread t2 = new Thread(t); 1515 t1.start(); 1616 t2.start(); 1717 } 1818 } 1919 2020 class TestReentrantlock extends Thread{ 2121 // 定义一个公平锁 2222 private static ReentrantLock lock = new ReentrantLock(true); 2323 public void run(){ 2424 for(int i = 0; i < 5; i++){ 2525 lock.lock(); 2626 try{ 2727 System.out.println(Thread.currentThread().getName() + " get lock"); 2828 }finally{ 2929 lock.unlock(); 3030 } 3131 } 3232 } 3333 3434 } 3535 3636 class TestSync extends Thread{ 3737 public void run(){ 3838 for(int i = 0; i < 5; i++){ 3939 synchronized (this) { 4040 System.out.println(Thread.currentThread().getName() + " get lock in TestSync"); 4141 } 4242 } 4343 } 4444 }
公平锁
1 1 /** 2 2 * 生产者消费者 3 3 * 重入锁&条件 4 4 * 条件 - Condition， 为Lock增加条件。当条件满足时，做什么事情，如加锁或解锁。如等待或唤醒 5 5 */ 6 6 package concurrent.t04; 7 7 8 8 import java.io.IOException; 9 9 import java.util.LinkedList; 10 10 import java.util.concurrent.TimeUnit; 11 11 import java.util.concurrent.locks.Condition; 12 12 import java.util.concurrent.locks.Lock; 13 13 import java.util.concurrent.locks.ReentrantLock; 14 14 15 15 public class TestContainer02<E> { 16 16 17 17 private final LinkedList<E> list = new LinkedList<>(); 18 18 private final int MAX = 10; 19 19 private int count = 0; 20 20 21 21 private Lock lock = new ReentrantLock(); 22 22 private Condition producer = lock.newCondition(); 23 23 private Condition consumer = lock.newCondition(); 24 24 25 25 public int getCount(){ 26 26 return count; 27 27 } 28 28 29 29 public void put(E e){ 30 30 lock.lock(); 31 31 try { 32 32 while(list.size() == MAX){ 33 33 System.out.println(Thread.currentThread().getName() + " 等待。。。"); 34 34 // 进入等待队列。释放锁标记。 35 35 // 借助条件，进入的等待队列。 36 36 producer.await(); 37 37 } 38 38 System.out.println(Thread.currentThread().getName() + " put 。。。"); 39 39 list.add(e); 40 40 count++; 41 41 // 借助条件，唤醒所有的消费者。 42 42 consumer.signalAll(); 43 43 } catch (InterruptedException e1) { 44 44 e1.printStackTrace(); 45 45 } finally { 46 46 lock.unlock(); 47 47 } 48 48 } 49 49 50 50 public E get(){ 51 51 E e = null; 52 52 53 53 lock.lock(); 54 54 try { 55 55 while(list.size() == 0){ 56 56 System.out.println(Thread.currentThread().getName() + " 等待。。。"); 57 57 // 借助条件，消费者进入等待队列 58 58 consumer.await(); 59 59 } 60 60 System.out.println(Thread.currentThread().getName() + " get 。。。"); 61 61 e = list.removeFirst(); 62 62 count--; 63 63 // 借助条件，唤醒所有的生产者 64 64 producer.signalAll(); 65 65 } catch (InterruptedException e1) { 66 66 e1.printStackTrace(); 67 67 } finally { 68 68 lock.unlock(); 69 69 } 70 70 71 71 return e; 72 72 } 73 73 74 74 public static void main(String[] args) { 75 75 final TestContainer02<String> c = new TestContainer02<>(); 76 76 for(int i = 0; i < 10; i++){ 77 77 new Thread(new Runnable() { 78 78 @Override 79 79 public void run() { 80 80 for(int j = 0; j < 5; j++){ 81 81 System.out.println(c.get()); 82 82 } 83 83 } 84 84 }, "consumer"+i).start(); 85 85 } 86 86 try { 87 87 TimeUnit.SECONDS.sleep(2); 88 88 } catch (InterruptedException e1) { 89 89 e1.printStackTrace(); 90 90 } 91 91 for(int i = 0; i < 2; i++){ 92 92 new Thread(new Runnable() { 93 93 @Override 94 94 public void run() { 95 95 for(int j = 0; j < 25; j++){ 96 96 c.put("container value " + j); 97 97 } 98 98 } 99 99 }, "producer"+i).start(); 100100 } 101101 } 102102 103103 }
重入锁&条件
1.8 ThreadLocal
ThreadLocal 提供了线程本地的实例。它与普通变量的区别在于，每个使用该变量的线程都会初始化一个完全独立的实例副本。ThreadLocal 变量通常被private static修饰。当一个线程结束时，它所使用的所有 ThreadLocal 相对的实例副本都可被回收。
ThreadLocal 适用于每个线程需要自己独立的实例且该实例需要在多个方法中被使用，也即变量在线程间隔离而在方法或类间共享的场景。每个 Thread 有自己的实例副本，且其它 Thread 不可访问，那就不存在多线程间共享的问题。
1 1 /** 2 2 * ThreadLocal 3 3 * 就是一个Map。key - 》 Thread.getCurrentThread(). value - 》 线程需要保存的变量。 4 4 * ThreadLocal.set(value) -> map.put(Thread.getCurrentThread(), value); 5 5 * ThreadLocal.get() -> map.get(Thread.getCurrentThread()); 6 6 * 内存问题 ： 在并发量高的时候，可能有内存溢出。 7 7 * 使用ThreadLocal的时候，一定注意回收资源问题，每个线程结束之前，将当前线程保存的线程变量一定要删除 。 8 8 * ThreadLocal.remove(); 9 9 */ 1010 package concurrent.t05; 1111 1212 import java.util.concurrent.TimeUnit; 1313 1414 public class Test_01 { 1515 1616 volatile static String name = "zhangsan"; 1717 static ThreadLocal<String> tl = new ThreadLocal<>(); 1818 1919 public static void main(String[] args) { 2020 new Thread(new Runnable() { 2121 @Override 2222 public void run() { 2323 try { 2424 TimeUnit.SECONDS.sleep(2); 2525 } catch (InterruptedException e) { 2626 e.printStackTrace(); 2727 } 2828 System.out.println(name); 2929 System.out.println(tl.get()); 3030 } 3131 }).start(); 3232 3333 new Thread(new Runnable() { 3434 @Override 3535 public void run() { 3636 try { 3737 TimeUnit.SECONDS.sleep(1); 3838 } catch (InterruptedException e) { 3939 e.printStackTrace(); 4040 } 4141 name = "lisi"; 4242 tl.set("wangwu"); 4343 } 4444 }).start(); 4545 } 4646 4747 }
ThreadLocal
如果调用 ThreadLocal 的 set 方法将一个对象放入Thread中的成员变量threadLocals 中，那么这个对象是永远不会被回收的，因为这个对象永远都被Thread中的成员变量threadLocals引用着，可能会造成 OutOfMemoryError。需要调用 ThreadLocal 的 remove 方法 将对象从thread中的成员变量threadLocals中删除掉。
二、同步容器
线程安全的容器对象： Vector, Hashtable。线程安全容器对象，都是使用 synchronized方法实现的。
concurrent 包中的同步容器，大多数是使用系统底层技术实现的线程安全。类似 native。Java8 中使用 CAS。
2.1 Map/Set
- ConcurrentHashMap/ConcurrentHashSet
底层哈希实现的同步 Map(Set)。效率高，线程安全。使用系统底层技术实现线程安全。量级较 synchronized 低。key 和 value 不能为 null。
- ConcurrentSkipListMap/ConcurrentSkipListSet
底层跳表（SkipList）实现的同步 Map(Set)。有序，效率比 ConcurrentHashMap 稍低。
1 1 /** 2 2 * 并发容器 - ConcurrentMap 3 3 */ 4 4 package concurrent.t06; 5 5 6 6 import java.util.HashMap; 7 7 import java.util.Hashtable; 8 8 import java.util.Map; 9 9 import java.util.Random; 1010 import java.util.concurrent.ConcurrentHashMap; 1111 import java.util.concurrent.ConcurrentSkipListMap; 1212 import java.util.concurrent.CountDownLatch; 1313 1414 public class Test_01_ConcurrentMap { 1515 1616 public static void main(String[] args) { 1717 final Map<String, String> map = new Hashtable<>(); 1818 // final Map<String, String> map = new ConcurrentHashMap<>(); 1919 // final Map<String, String> map = new ConcurrentSkipListMap<>(); 2020 final Random r = new Random(); 2121 Thread[] array = new Thread[100]; 2222 final CountDownLatch latch = new CountDownLatch(array.length); 2323 2424 long begin = System.currentTimeMillis(); 2525 for(int i = 0; i < array.length; i++){ 2626 array[i] = new Thread(new Runnable() { 2727 @Override 2828 public void run() { 2929 for(int j = 0; j < 10000; j++){ 3030 map.put("key"+r.nextInt(100000), "value"+r.nextInt(100000)); 3131 } 3232 latch.countDown(); 3333 } 3434 }); 3535 } 3636 for(Thread t : array){ 3737 t.start(); 3838 } 3939 try { 4040 latch.await(); 4141 } catch (InterruptedException e) { 4242 e.printStackTrace(); 4343 } 4444 long end = System.currentTimeMillis(); 4545 System.out.println("执行时间为 ： " + (end-begin) + "毫秒！"); 4646 } 4747 4848 }
并发容器 - ConcurrentMap
2.2 List
- CopyOnWriteArrayList
1 1 /** 2 2 * 并发容器 - CopyOnWriteList 3 3 * 写时复制集合。写入效率低，读取效率高。每次写入数据，都会创建一个新的底层数组。 4 4 */ 5 5 package concurrent.t06; 6 6 7 7 import java.util.ArrayList; 8 8 import java.util.List; 9 9 import java.util.Random; 1010 import java.util.Vector; 1111 import java.util.concurrent.CopyOnWriteArrayList; 1212 import java.util.concurrent.CountDownLatch; 1313 1414 public class Test_02_CopyOnWriteList { 1515 1616 public static void main(String[] args) { 1717 // final List<String> list = new ArrayList<>(); 1818 // final List<String> list = new Vector<>(); 1919 final List<String> list = new CopyOnWriteArrayList<>(); 2020 final Random r = new Random(); 2121 Thread[] array = new Thread[100]; 2222 final CountDownLatch latch = new CountDownLatch(array.length); 2323 2424 long begin = System.currentTimeMillis(); 2525 for(int i = 0; i < array.length; i++){ 2626 array[i] = new Thread(new Runnable() { 2727 @Override 2828 public void run() { 2929 for(int j = 0; j < 1000; j++){ 3030 list.add("value" + r.nextInt(100000)); 3131 } 3232 latch.countDown(); 3333 } 3434 }); 3535 } 3636 for(Thread t : array){ 3737 t.start(); 3838 } 3939 try { 4040 latch.await(); 4141 } catch (InterruptedException e) { 4242 e.printStackTrace(); 4343 } 4444 long end = System.currentTimeMillis(); 4545 System.out.println("执行时间为 ： " + (end-begin) + "毫秒！"); 4646 System.out.println("List.size() : " + list.size()); 4747 } 4848 4949 }
CopyOnWriteList
2.3 Queue
- ConcurrentLinkedQueue 基础链表同步队列。
1 1 /** 2 2 * 并发容器 - ConcurrentLinkedQueue 3 3 * 队列 - 链表实现的。 4 4 */ 5 5 package concurrent.t06; 6 6 7 7 import java.util.Queue; 8 8 import java.util.concurrent.ConcurrentLinkedQueue; 9 9 1010 public class Test_03_ConcurrentLinkedQueue { 1111 1212 public static void main(String[] args) { 1313 Queue<String> queue = new ConcurrentLinkedQueue<>(); 1414 for(int i = 0; i < 10; i++){ 1515 queue.offer("value" + i); 1616 } 1717 1818 System.out.println(queue); 1919 System.out.println(queue.size()); 2020 2121 // peek() -> 查看queue中的首数据 2222 System.out.println(queue.peek()); 2323 System.out.println(queue.size()); 2424 2525 // poll() -> 获取queue中的首数据 2626 System.out.println(queue.poll()); 2727 System.out.println(queue.size()); 2828 } 2929 3030 }
ConcurrentLinkedQueue
- LinkedBlockingQueue 阻塞队列，队列容量不足自动阻塞，队列容量为 0 自动阻塞。
1 1 /** 2 2 * 并发容器 - LinkedBlockingQueue 3 3 * 阻塞容器。 4 4 * put & take - 自动阻塞。 5 5 * put自动阻塞， 队列容量满后，自动阻塞 6 6 * take自动阻塞方法， 队列容量为0后，自动阻塞。 7 7 */ 8 8 package concurrent.t06; 9 9 1010 import java.util.Random; 1111 import java.util.concurrent.BlockingQueue; 1212 import java.util.concurrent.LinkedBlockingQueue; 1313 import java.util.concurrent.TimeUnit; 1414 1515 public class Test_04_LinkedBlockingQueue { 1616 1717 final BlockingQueue<String> queue = new LinkedBlockingQueue<>(); 1818 final Random r = new Random(); 1919 2020 public static void main(String[] args) { 2121 final Test_04_LinkedBlockingQueue t = new Test_04_LinkedBlockingQueue(); 2222 2323 new Thread(new Runnable() { 2424 @Override 2525 public void run() { 2626 while(true){ 2727 try { 2828 t.queue.put("value"+t.r.nextInt(1000)); 2929 TimeUnit.SECONDS.sleep(1); 3030 } catch (InterruptedException e) { 3131 e.printStackTrace(); 3232 } 3333 } 3434 } 3535 }, "producer").start(); 3636 3737 for(int i = 0; i < 3; i++){ 3838 new Thread(new Runnable() { 3939 @Override 4040 public void run() { 4141 while(true){ 4242 try { 4343 System.out.println(Thread.currentThread().getName() + 4444 " - " + t.queue.take()); 4545 } catch (InterruptedException e) { 4646 e.printStackTrace(); 4747 } 4848 } 4949 } 5050 }, "consumer"+i).start(); 5151 } 5252 } 5353 5454 }
LinkedBlockingQueue
- ArrayBlockingQueue 底层数组实现的有界队列
1 1 /** 2 2 * 并发容器 - ArrayBlockingQueue 3 3 * 有界容器。 4 4 * 当容量不足的时候，有阻塞能力。 5 5 *add 方法在容量不足的时候，抛出异常。 6 6 *put 方法在容量不足的时候，阻塞等待。 7 7 *offer 方法， 8 8 *单参数 offer 方法，不阻塞。容量不足的时候，返回 false。当前新增数据操作放弃。 9 9 *三参数 offer 方法（offer(value,times,timeunit)），容量不足的时候，阻塞 times 时长（单 1010 *位为 timeunit），如果在阻塞时长内，有容量空闲，新增数据返回 true。如果阻塞时长范围 1111 *内，无容量空闲，放弃新增数据，返回 false。 1212 */ 1313 package concurrent.t06; 1414 1515 import java.util.concurrent.ArrayBlockingQueue; 1616 import java.util.concurrent.BlockingQueue; 1717 import java.util.concurrent.TimeUnit; 1818 1919 public class Test_05_ArrayBlockingQueue { 2020 2121 final BlockingQueue<String> queue = new ArrayBlockingQueue<>(3); 2222 2323 public static void main(String[] args) { 2424 final Test_05_ArrayBlockingQueue t = new Test_05_ArrayBlockingQueue(); 2525 2626 for(int i = 0; i < 5; i++){ 2727 // System.out.println("add method : " + t.queue.add("value"+i)); 2828 /*try { 2929 t.queue.put("put"+i); 3030 } catch (InterruptedException e) { 3131 e.printStackTrace(); 3232 } 3333 System.out.println("put method : " + i);*/ 3434 // System.out.println("offer method : " + t.queue.offer("value"+i)); 3535 try { 3636 System.out.println("offer method : " + 3737 t.queue.offer("value"+i, 1, TimeUnit.SECONDS)); 3838 } catch (InterruptedException e) { 3939 e.printStackTrace(); 4040 } 4141 } 4242 4343 System.out.println(t.queue); 4444 } 4545 4646 }
ArrayBlockingQueue
- DelayQueue 延时队列。根据比较机制，实现自定义处理顺序的队列。常用于定时任务。
1 1 /** 2 2 * 并发容器 - DelayQueue 3 3 * 无界容器。 4 4 */ 5 5 package concurrent.t06; 6 6 7 7 import java.util.concurrent.BlockingQueue; 8 8 import java.util.concurrent.DelayQueue; 9 9 import java.util.concurrent.Delayed; 1010 import java.util.concurrent.TimeUnit; 1111 1212 public class Test_06_DelayQueue { 1313 1414 static BlockingQueue<MyTask_06> queue = new DelayQueue<>(); 1515 1616 public static void main(String[] args) throws InterruptedException { 1717 long value = System.currentTimeMillis(); 1818 MyTask_06 task1 = new MyTask_06(value + 2000); 1919 MyTask_06 task2 = new MyTask_06(value + 1000); 2020 MyTask_06 task3 = new MyTask_06(value + 3000); 2121 MyTask_06 task4 = new MyTask_06(value + 2500); 2222 MyTask_06 task5 = new MyTask_06(value + 1500); 2323 2424 queue.put(task1); 2525 queue.put(task2); 2626 queue.put(task3); 2727 queue.put(task4); 2828 queue.put(task5); 2929 3030 System.out.println(queue); 3131 System.out.println(value); 3232 for(int i = 0; i < 5; i++){ 3333 System.out.println(queue.take()); 3434 } 3535 } 3636 3737 } 3838 3939 class MyTask_06 implements Delayed { 4040 4141 private long compareValue; 4242 4343 public MyTask_06(long compareValue){ 4444 this.compareValue = compareValue; 4545 } 4646 4747 /** 4848 * 比较大小。自动实现升序 4949 * 建议和getDelay方法配合完成。 5050 * 如果在DelayQueue是需要按时间完成的计划任务，必须配合getDelay方法完成。 5151 */ 5252 @Override 5353 public int compareTo(Delayed o) { 5454 return (int)(this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS)); 5555 } 5656 5757 /** 5858 * 获取计划时长的方法。 5959 * 根据参数TimeUnit来决定，如何返回结果值。 6060 */ 6161 @Override 6262 public long getDelay(TimeUnit unit) { 6363 return unit.convert(compareValue - System.currentTimeMillis(), TimeUnit.MILLISECONDS); 6464 } 6565 6666 @Override 6767 public String toString(){ 6868 return "Task compare value is : " + this.compareValue; 6969 } 7070 7171 }
DelayQueue
- LinkedTransferQueue 转移队列，使用 transfer 方法，实现数据的即时处理。没有消费者，就阻塞。
1 1 /** 2 2 * 并发容器 - LinkedTransferQueue 3 3 * 转移队列 4 4 * add - 队列会保存数据，不做阻塞等待。 5 5 * transfer - 是TransferQueue的特有方法。必须有消费者（take()方法的调用者）。 6 6 * 如果没有任意线程消费数据，transfer方法阻塞。一般用于处理即时消息。 7 7 */ 8 8 package concurrent.t06; 9 9 1010 import java.util.concurrent.LinkedTransferQueue; 1111 import java.util.concurrent.TimeUnit; 1212 import java.util.concurrent.TransferQueue; 1313 1414 public class Test_07_TransferQueue { 1515 1616 TransferQueue<String> queue = new LinkedTransferQueue<>(); 1717 1818 public static void main(String[] args) { 1919 final Test_07_TransferQueue t = new Test_07_TransferQueue(); 2020 2121 /*new Thread(new Runnable() { 2222 @Override 2323 public void run() { 2424 try { 2525 System.out.println(Thread.currentThread().getName() + " thread begin " ); 2626 System.out.println(Thread.currentThread().getName() + " - " + t.queue.take()); 2727 } catch (InterruptedException e) { 2828 e.printStackTrace(); 2929 } 3030 } 3131 }, "output thread").start(); 3232 3333 try { 3434 TimeUnit.SECONDS.sleep(2); 3535 } catch (InterruptedException e) { 3636 e.printStackTrace(); 3737 } 3838 3939 try { 4040 t.queue.transfer("test string"); 4141 } catch (InterruptedException e) { 4242 e.printStackTrace(); 4343 }*/ 4444 4545 new Thread(new Runnable() { 4646 4747 @Override 4848 public void run() { 4949 try { 5050 t.queue.transfer("test string"); 5151 // t.queue.add("test string"); 5252 System.out.println("add ok"); 5353 } catch (Exception e) { 5454 e.printStackTrace(); 5555 } 5656 } 5757 }).start(); 5858 5959 try { 6060 TimeUnit.SECONDS.sleep(2); 6161 } catch (InterruptedException e) { 6262 e.printStackTrace(); 6363 } 6464 6565 new Thread(new Runnable() { 6666 @Override 6767 public void run() { 6868 try { 6969 System.out.println(Thread.currentThread().getName() + " thread begin " ); 7070 System.out.println(Thread.currentThread().getName() + " - " + t.queue.take()); 7171 } catch (InterruptedException e) { 7272 e.printStackTrace(); 7373 } 7474 } 7575 }, "output thread").start(); 7676 7777 } 7878 7979 }
LinkedTransferQueue
- SynchronusQueue 同步队列，是一个容量为 0 的队列。是一个特殊的 TransferQueue。必须现有消费线程等待，才能使用的队列。
1 1 /** 2 2 * 并发容器 - SynchronousQueue 3 3 * 必须现有消费线程等待，才能使用的队列。 4 4 * add 方法，无阻塞。若没有消费线程阻塞等待数据，则抛出异常。 5 5 * put 方法，有阻塞。若没有消费线程阻塞等待数据，则阻塞。 6 6 */ 7 7 package concurrent.t06; 8 8 9 9 import java.util.concurrent.BlockingQueue; 1010 import java.util.concurrent.SynchronousQueue; 1111 import java.util.concurrent.TimeUnit; 1212 1313 public class Test_08_SynchronusQueue { 1414 1515 BlockingQueue<String> queue = new SynchronousQueue<>(); 1616 1717 public static void main(String[] args) { 1818 final Test_08_SynchronusQueue t = new Test_08_SynchronusQueue(); 1919 2020 new Thread(new Runnable() { 2121 @Override 2222 public void run() { 2323 try { 2424 System.out.println(Thread.currentThread().getName() + " thread begin " ); 2525 try { 2626 TimeUnit.SECONDS.sleep(2); 2727 } catch (InterruptedException e) { 2828 e.printStackTrace(); 2929 } 3030 System.out.println(Thread.currentThread().getName() + " - " + t.queue.take()); 3131 } catch (InterruptedException e) { 3232 e.printStackTrace(); 3333 } 3434 } 3535 }, "output thread").start(); 3636 3737 /*try { 3838 TimeUnit.SECONDS.sleep(3); 3939 } catch (InterruptedException e) { 4040 e.printStackTrace(); 4141 }*/ 4242 // t.queue.add("test add"); 4343 try { 4444 t.queue.put("test put"); 4545 } catch (InterruptedException e) { 4646 e.printStackTrace(); 4747 } 4848 4949 System.out.println(Thread.currentThread().getName() + " queue size : " + t.queue.size()); 5050 } 5151 5252 }
SynchronousQueue
三、 ThreadPool&Executor
3.1 Executor
线程池顶级接口。定义方法，void execute(Runnable)。方法是用于处理任务的一个服务方法。调用者提供 Runnable 接口的实现，线程池通过线程执行这个 Runnable。服务方法无返回值的。是 Runnable 接口中的 run 方法无返回值。
常用方法 - void execute(Runnable)
作用是： 启动线程任务的。
1 1 /** 2 2 * 线程池 3 3 * Executor - 线程池底层处理机制。 4 4 * 在使用线程池的时候，底层如何调用线程中的逻辑。 5 5 */ 6 6 package concurrent.t08; 7 7 8 8 import java.util.concurrent.Executor; 9 9 1010 public class Test_01_MyExecutor implements Executor { 1111 public static void main(String[] args) { 1212 new Test_01_MyExecutor().execute(new Runnable() { 1313 @Override 1414 public void run() { 1515 System.out.println(Thread.currentThread().getName() + " - test executor"); 1616 } 1717 }); 1818 } 1919 2020 @Override 2121 public void execute(Runnable command) { 2222 new Thread(command).start(); 2323 } 2424 }
线程池底层处理机制
3.2 ExecutorService
Executor 接口的子接口。提供了一个新的服务方法，submit。有返回值（Future 类型）。submit 方法提供了 overload 方法。其中有参数类型为 Runnable 的，不需要提供返回值的；有参数类型为 Callable，可以提供线程执行后的返回值。
Future，是 submit 方法的返回值。代表未来，也就是线程执行结束后的一种结果。如返回值。
常见方法 - void execute(Runnable)， Future submit(Callable)， Future submit(Runnable)
线程池状态： Running， ShuttingDown， Termitnaed
Running - 线程池正在执行中。活动状态。* ShuttingDown - 线程池正在关闭过程中。优雅关闭。一旦进入这个状态，线程池不再接收新的任务，处理所有已接收的任务，处理完毕后，关闭线程池。
- Terminated - 线程池已经关闭。
3.3 Future
未来结果，代表线程任务执行结束后的结果。获取线程执行结果的方式是通过 get 方法获取的。get 无参，阻塞等待线程执行结束，并得到结果。get 有参，阻塞固定时长，等待
线程执行结束后的结果，如果在阻塞时长范围内，线程未执行结束，抛出异常。
常用方法： T get() T get(long, TimeUnit)
1 1 /** 2 2 * 线程池 3 3 * 固定容量线程池 4 4 */ 5 5 package concurrent.t08; 6 6 7 7 import java.util.concurrent.Callable; 8 8 import java.util.concurrent.ExecutionException; 9 9 import java.util.concurrent.ExecutorService; 1010 import java.util.concurrent.Executors; 1111 import java.util.concurrent.Future; 1212 import java.util.concurrent.FutureTask; 1313 import java.util.concurrent.TimeUnit; 1414 1515 public class Test_03_Future { 1616 1717 public static void main(String[] args) throws InterruptedException, ExecutionException { 1818 /*FutureTask<String> task = new FutureTask<>(new Callable<String>() { 1919 @Override 2020 public String call() throws Exception { 2121 return "first future task"; 2222 } 2323 }); 2424 2525 new Thread(task).start(); 2626 2727 System.out.println(task.get());*/ 2828 2929 ExecutorService service = Executors.newFixedThreadPool(1); 3030 3131 Future<String> future = service.submit(new Callable<String>() { 3232 @Override 3333 public String call() { 3434 try { 3535 TimeUnit.MILLISECONDS.sleep(500); 3636 } catch (InterruptedException e) { 3737 e.printStackTrace(); 3838 } 3939 System.out.println("aaa"); 4040 return Thread.currentThread().getName() + " - test executor"; 4141 } 4242 }); 4343 System.out.println(future); 4444 System.out.println(future.isDone()); // 查看线程是否结束， 任务是否完成。 call方法是否执行结束 4545 4646 System.out.println(future.get()); // 获取call方法的返回值。 4747 System.out.println(future.isDone()); 4848 } 4949 5050 }
Future
3.4 Callable
可执行接口。 类似 Runnable 接口。也是可以启动一个线程的接口。其中定义的方法是call。call 方法的作用和 Runnable 中的 run 方法完全一致。call 方法有返回值。
接口方法 ： Object call();相当于 Runnable 接口中的 run 方法。区别为此方法有返回值。不能抛出已检查异常。
和 Runnable 接口的选择 - 需要返回值或需要抛出异常时，使用 Callable，其他情况可任意选择。
3.5 Executors
工具类型。为 Executor 线程池提供工具方法。可以快速的提供若干种线程池。如：固定容量的，无限容量的，容量为 1 等各种线程池。
线程池是一个进程级的重量级资源。默认的生命周期和 JVM 一致。当开启线程池后，直到 JVM 关闭为止，是线程池的默认生命周期。如果手工调用 shutdown 方法，那么线程池执行所有的任务后，自动关闭。
开始 - 创建线程池。
结束 - JVM 关闭或调用 shutdown 并处理完所有的任务。
类似 Arrays，Collections 等工具类型的功用。
3.6 FixedThreadPool
容量固定的线程池。活动状态和线程池容量是有上限的线程池。所有的线程池中，都有一个任务队列。使用的是 BlockingQueue<Runnable>作为任务的载体。当任务数量大于线程池容量的时候，没有运行的任务保存在任务队列中，当线程有空闲的，自动从队列中取出任务执行。
使用场景： 大多数情况下，使用的线程池，首选推荐 FixedThreadPool。OS 系统和硬件是有线程支持上限。不能随意的无限制提供线程池。
线程池默认的容量上限是 Integer.MAX_VALUE。
常见的线程池容量： PC - 200。 服务器 - 1000~10000
queued tasks - 任务队列
completed tasks - 结束任务队列
1 1 /** 2 2 * 线程池 3 3 * 固定容量线程池 4 4 * FixedThreadPool - 固定容量线程池。创建线程池的时候，容量固定。 5 5 * 构造的时候，提供线程池最大容量 6 6 * new xxxxx -> 7 7 * ExecutorService - 线程池服务类型。所有的线程池类型都实现这个接口。 8 8 * 实现这个接口，代表可以提供线程池能力。 9 9 * shutdown - 优雅关闭。 不是强行关闭线程池，回收线程池中的资源。而是不再处理新的任务，将已接收的任务处理完毕后 1010 * 再关闭。 1111 * Executors - Executor的工具类。类似Collection和Collections的关系。 1212 * 可以更简单的创建若干种线程池。 1313 */ 1414 package concurrent.t08; 1515 1616 import java.util.concurrent.ExecutorService; 1717 import java.util.concurrent.Executors; 1818 import java.util.concurrent.TimeUnit; 1919 2020 public class Test_02_FixedThreadPool { 2121 2222 public static void main(String[] args) { 2323 ExecutorService service = 2424 Executors.newFixedThreadPool(5); 2525 for(int i = 0; i < 6; i++){ 2626 service.execute(new Runnable() { 2727 @Override 2828 public void run() { 2929 try { 3030 TimeUnit.MILLISECONDS.sleep(500); 3131 } catch (InterruptedException e) { 3232 e.printStackTrace(); 3333 } 3434 System.out.println(Thread.currentThread().getName() + " - test executor"); 3535 } 3636 }); 3737 } 3838 3939 System.out.println(service); 4040 4141 service.shutdown(); 4242 // 是否已经结束， 相当于回收了资源。 4343 System.out.println(service.isTerminated()); 4444 // 是否已经关闭， 是否调用过shutdown方法 4545 System.out.println(service.isShutdown()); 4646 System.out.println(service); 4747 4848 try { 4949 TimeUnit.SECONDS.sleep(2); 5050 } catch (InterruptedException e) { 5151 e.printStackTrace(); 5252 } 5353 5454 // service.shutdown(); 5555 System.out.println(service.isTerminated()); 5656 System.out.println(service.isShutdown()); 5757 System.out.println(service); 5858 } 5959 6060 }
FixedThreadPool
3.7 CachedThreadPool
缓存的线程池。容量不限（Integer.MAX_VALUE）。自动扩容。容量管理策略：如果线程池中的线程数量不满足任务执行，创建新的线程。每次有新任务无法即时处理的时候，都会创建新的线程。当线程池中的线程空闲时长达到一定的临界值（默认 60 秒），自动释放线程。默认线程空闲 60 秒，自动销毁。
应用场景： 内部应用或测试应用。 内部应用，有条件的内部数据瞬间处理时应用，如：
电信平台夜间执行数据整理（有把握在短时间内处理完所有工作，且对硬件和软件有足够的信心）。 测试应用，在测试的时候，尝试得到硬件或软件的最高负载量，用于提供FixedThreadPool 容量的指导。
1 1 /** 2 2 * 线程池 3 3 * 无容量限制的线程池（最大容量默认为Integer.MAX_VALUE） 4 4 */ 5 5 package concurrent.t08; 6 6 7 7 import java.util.concurrent.ExecutorService; 8 8 import java.util.concurrent.Executors; 9 9 import java.util.concurrent.TimeUnit; 1010 1111 public class Test_05_CachedThreadPool { 1212 1313 public static void main(String[] args) { 1414 ExecutorService service = Executors.newCachedThreadPool(); 1515 System.out.println(service); 1616 1717 for(int i = 0; i < 5; i++){ 1818 service.execute(new Runnable() { 1919 @Override 2020 public void run() { 2121 try { 2222 TimeUnit.MILLISECONDS.sleep(500); 2323 } catch (InterruptedException e) { 2424 e.printStackTrace(); 2525 } 2626 System.out.println(Thread.currentThread().getName() + " - test executor"); 2727 } 2828 }); 2929 } 3030 3131 System.out.println(service); 3232 3333 try { 3434 TimeUnit.SECONDS.sleep(65); 3535 } catch (InterruptedException e) { 3636 e.printStackTrace(); 3737 } 3838 3939 System.out.println(service); 4040 } 4141 4242 }
CachedThreadPool
3.8 ScheduledThreadPool
计划任务线程池。可以根据计划自动执行任务的线程池。
scheduleAtFixedRate(Runnable, start_limit, limit, timeunit)
runnable - 要执行的任务。
start_limit - 第一次任务执行的间隔。
limit - 多次任务执行的间隔。
timeunit - 多次任务执行间隔的时间单位。
使用场景： 计划任务时选用（DelaydQueue），如：电信行业中的数据整理，没分钟整理，没消失整理，每天整理等。
1 1 /** 2 2 * 线程池 3 3 * 计划任务线程池。 4 4 */ 5 5 package concurrent.t08; 6 6 7 7 import java.util.concurrent.Executors; 8 8 import java.util.concurrent.ScheduledExecutorService; 9 9 import java.util.concurrent.TimeUnit; 1010 1111 public class Test_07_ScheduledThreadPool { 1212 1313 public static void main(String[] args) { 1414 ScheduledExecutorService service = Executors.newScheduledThreadPool(3); 1515 System.out.println(service); 1616 1717 // 定时完成任务。 scheduleAtFixedRate(Runnable, start_limit, limit, timeunit) 1818 // runnable - 要执行的任务。 1919 service.scheduleAtFixedRate(new Runnable() { 2020 @Override 2121 public void run() { 2222 try { 2323 TimeUnit.MILLISECONDS.sleep(500); 2424 } catch (InterruptedException e) { 2525 e.printStackTrace(); 2626 } 2727 System.out.println(Thread.currentThread().getName()); 2828 } 2929 }, 0, 300, TimeUnit.MILLISECONDS); 3030 3131 } 3232 3333 }
ScheduledThreadPool
3.9 SingleThreadExceutor
单一容量的线程池。使用场景： 保证任务顺序时使用。如： 游戏大厅中的公共频道聊天。秒杀。
1 1 /** 2 2 * 线程池 3 3 * 容量为1的线程池。 顺序执行。 4 4 */ 5 5 package concurrent.t08; 6 6 7 7 import java.util.concurrent.ExecutorService; 8 8 import java.util.concurrent.Executors; 9 9 import java.util.concurrent.TimeUnit; 1010 1111 public class Test_06_SingleThreadExecutor { 1212 1313 public static void main(String[] args) { 1414 ExecutorService service = Executors.newSingleThreadExecutor(); 1515 System.out.println(service); 1616 1717 for(int i = 0; i < 5; i++){ 1818 service.execute(new Runnable() { 1919 @Override 2020 public void run() { 2121 try { 2222 TimeUnit.MILLISECONDS.sleep(500); 2323 } catch (InterruptedException e) { 2424 e.printStackTrace(); 2525 } 2626 System.out.println(Thread.currentThread().getName() + " - test executor"); 2727 } 2828 }); 2929 } 3030 3131 } 3232 3333 }
SingleThreadExecutor
3.10 ForkJoinPool
分支合并线程池（mapduce 类似的设计思想）。适合用于处理复杂任务。
初始化线程容量与 CPU 核心数相关。
线程池中运行的内容必须是 ForkJoinTask 的子类型（RecursiveTask,RecursiveAction）。ForkJoinPool - 分支合并线程池。 可以递归完成复杂任务。要求可分支合并的任务必须是 ForkJoinTask 类型的子类型。其中提供了分支和合并的能力。ForkJoinTask 类型提供了两个抽象子类型，RecursiveTask 有返回结果的分支合并任务,RecursiveAction 无返回结果的分支合并任务。（Callable/Runnable）compute 方法：就是任务的执行逻辑。
ForkJoinPool 没有所谓的容量。默认都是 1 个线程。根据任务自动的分支新的子线程。当子线程任务结束后，自动合并。所谓自动是根据 fork 和 join 两个方法实现的。
应用： 主要是做科学计算或天文计算的。数据分析的。
1 1 /** 2 2 * 线程池 3 3 * 分支合并线程池。 4 4 */ 5 5 package concurrent.t08; 6 6 7 7 import java.io.IOException; 8 8 import java.util.Random; 9 9 import java.util.concurrent.ExecutionException; 1010 import java.util.concurrent.ForkJoinPool; 1111 import java.util.concurrent.Future; 1212 import java.util.concurrent.RecursiveTask; 1313 1414 public class Test_08_ForkJoinPool { 1515 1616 final static int[] numbers = new int[1000000]; 1717 final static int MAX_SIZE = 50000; 1818 final static Random r = new Random(); 1919 2020 2121 static{ 2222 for(int i = 0; i < numbers.length; i++){ 2323 numbers[i] = r.nextInt(1000); 2424 } 2525 } 2626 2727 static class AddTask extends RecursiveTask<Long>{ // RecursiveAction 2828 int begin, end; 2929 public AddTask(int begin, int end){ 3030 this.begin = begin; 3131 this.end = end; 3232 } 3333 3434 // 3535 protected Long compute(){ 3636 if((end - begin) < MAX_SIZE){ 3737 long sum = 0L; 3838 for(int i = begin; i < end; i++){ 3939 sum += numbers[i]; 4040 } 4141 // System.out.println("form " + begin + " to " + end + " sum is : " + sum); 4242 return sum; 4343 }else{ 4444 int middle = begin + (end - begin)/2; 4545 AddTask task1 = new AddTask(begin, middle); 4646 AddTask task2 = new AddTask(middle, end); 4747 task1.fork();// 就是用于开启新的任务的。 就是分支工作的。 就是开启一个新的线程任务。 4848 task2.fork(); 4949 // join - 合并。将任务的结果获取。 这是一个阻塞方法。一定会得到结果数据。 5050 return task1.join() + task2.join(); 5151 } 5252 } 5353 } 5454 5555 public static void main(String[] args) throws InterruptedException, ExecutionException, IOException { 5656 long result = 0L; 5757 for(int i = 0; i < numbers.length; i++){ 5858 result += numbers[i]; 5959 } 6060 System.out.println(result); 6161 6262 ForkJoinPool pool = new ForkJoinPool(); 6363 AddTask task = new AddTask(0, numbers.length); 6464 6565 Future<Long> future = pool.submit(task); 6666 System.out.println(future.get()); 6767 6868 } 6969 7070 }
ForkJoinPool
3.11 ThreadPoolExecutor
线程池底层实现。除 ForkJoinPool 外，其他常用线程池底层都是使用 ThreadPoolExecutor实现的。
1 public ThreadPoolExecutor 2 (int corePoolSize, // 核心容量，创建线程池的时候，默认有多少线程。也是线程池保持的最少线程数 3 int maximumPoolSize, // 最大容量，线程池最多有多少线程 4 long keepAliveTime, // 生命周期，0 为永久。当线程空闲多久后，自动回收。 5 TimeUnit unit, // 生命周期单位，为生命周期提供单位，如：秒，毫秒 6 BlockingQueue<Runnable> workQueue // 任务队列，阻塞队列。注意，泛型必须是 7 Runnable 8 ); 9//使用场景： 默认提供的线程池不满足条件时使用。如：初始线程数据 4，最大线程数200，线程空闲周期 30 秒。
1 1 /** 2 2 * 线程池 3 3 * 固定容量线程池 4 4 */ 5 5 package concurrent.t08; 6 6 7 7 import java.util.ArrayList; 8 8 import java.util.concurrent.ExecutorService; 9 9 import java.util.concurrent.LinkedBlockingQueue; 1010 import java.util.concurrent.ThreadPoolExecutor; 1111 import java.util.concurrent.TimeUnit; 1212 1313 public class Test_09_ThreadPoolExecutor { 1414 1515 public static void main(String[] args) { 1616 // 模拟fixedThreadPool， 核心线程5个，最大容量5个，线程的生命周期无限。 1717 ExecutorService service = 1818 new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, 1919 new LinkedBlockingQueue<Runnable>()); 2020 2121 for(int i = 0; i < 6; i++){ 2222 service.execute(new Runnable() { 2323 @Override 2424 public void run() { 2525 try { 2626 TimeUnit.MILLISECONDS.sleep(500); 2727 } catch (InterruptedException e) { 2828 e.printStackTrace(); 2929 } 3030 System.out.println(Thread.currentThread().getName() + " - test executor"); 3131 } 3232 }); 3333 } 3434 3535 System.out.println(service); 3636 3737 service.shutdown(); 3838 System.out.println(service.isTerminated()); 3939 System.out.println(service.isShutdown()); 4040 System.out.println(service); 4141 4242 try { 4343 TimeUnit.SECONDS.sleep(2); 4444 } catch (InterruptedException e) { 4545 e.printStackTrace(); 4646 } 4747 4848 service.shutdown(); 4949 System.out.println(service.isTerminated()); 5050 System.out.println(service.isShutdown()); 5151 System.out.println(service); 5252 5353 } 5454 5555 }
ThreadPoolExecutor