【Java并发】四、JDK并发包

拼搏现实的明天。 2022-04-13 02:51 298阅读 0赞

# JDK并发包 #

### 文章目录 ###

*  JDK并发包
 *   *  同步控制
     *   *  重入锁ReentrantLock
         *  Condition
         *  信号量(Semaphore)
         *  ReadWriteLock 读写锁
         *  CountDownLatch
         *  循环栅栏(CyclicBarrier)
         *  线程阻塞工具类(LockSupport)
     *  线程池
     *   *  不要重复造轮子：JDK中的线程池
         *  线程池实现原理
         *  拒绝策略
         *  自定义ThreadFactory
         *  线程池扩展
         *  submit()与execute()
         *  合理选择线程池数量
         *  Fork/Join和ForkJoinPool
     *  并发容器

## 同步控制 ##

### 重入锁ReentrantLock ###

`ReentrantLock`可以完全替代`synchronized`，在JDK1.5中引入重入锁时，重入锁的性能远好于`synchronized`，但从JDK1.6开始对`synchronized`做了很多优化，现在两者的性能差距并不大。

public class ReentrantLockTest { 
    
        private static int sum = 0;
    
        public static void main(String[] args) throws InterruptedException { 
            ReentrantLock lock = new ReentrantLock();
            Thread[] threads = new Thread[100];
            for (int i = 1; i <= 100; i++) { 
                int finalI = i;
                Thread thread = new Thread(() -> { 
                    lock.lock();
                    try { 
                        sum += finalI;
                    } finally { 
                        lock.unlock();
                    }
                });
                threads[i-1] = thread;
                thread.start();
            }
    
            //等待线程执行完毕
            for (Thread thread : threads) { 
                thread.join();
            }
    
            System.out.println("sum = " + sum);
        }
    
    }

可以看出重入锁相较于`synchronized`有着显式的操作过程，加锁释放锁都需要程序猿手动指定，这使得重入锁对逻辑控制的灵活性远高于`synchronized`，但同时也必须注意：退出临界区的时候必须释放锁，否者其它线程将永远没有机会进入临界区。

同一个线程可以重复连续地获得同一个锁，即可以重复调用`lock.lock()`，但是相应的，退出临界区的时候必须释放相同次数的锁，如果次数少于加锁次数，其它线程一样再也不能进入临界区，如果次数多余加锁次数，将抛出`java.lang.IllegalMonitorStateException`，但锁是会被释放，其他线程可以获得锁。

Thread thread = new Thread(() -> { 
        lock.lock();
        lock.lock();
        try { 
            sum += finalI;
        } finally { 
            lock.unlock();
            lock.unlock();
            lock.unlock();//java.lang.IllegalMonitorStateException
        }
    });

重入锁的一些功能：

*  中断响应：如果使用`synchronized`，这个线程只有两种结果，一直等不到锁或者等到锁继续执行。重入锁提供了第三种可能：中断，假设一个线程占用了锁执行了太久的时间，另一个线程正在等待锁，可以直接中断另一个线程。
    
        public class ReentrantLockInterrupt { 
        
            public static void main(String[] args) throws InterruptedException { 
                ReentrantLock lock1 = new ReentrantLock();
                ReentrantLock lock2 = new ReentrantLock();
        
                Thread thread1 = new Thread(() -> { 
                    try { 
                        lock1.lockInterruptibly();
                        System.out.println("thread1 hold lock1...");
                        Thread.sleep(1000L);
                        lock2.lockInterruptibly();
                        System.out.println("thread1 hold lock2...");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }finally { 
                        if(lock1.isHeldByCurrentThread()){ 
                            lock1.unlock();
                            System.out.println("thread1 release lock1...");
                        }
                        if(lock2.isHeldByCurrentThread()){ 
                            lock2.unlock();
                            System.out.println("thread1 release lock2...");
                        }
                    }
                });
        
                Thread thread2 = new Thread(() -> { 
                    try { 
                        lock2.lockInterruptibly();
                        System.out.println("thread2 hold lock2...");
                        Thread.sleep(100L);
                        lock1.lockInterruptibly();
                        System.out.println("thread2 hold lock1...");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }finally { 
                        if(lock2.isHeldByCurrentThread()){ 
                            lock2.unlock();
                            System.out.println("thread2 release lock2...");
                        }
                        if(lock1.isHeldByCurrentThread()){ 
                            lock1.unlock();
                            System.out.println("thread2 release lock1...");
                        }
                    }
                });
        
                thread1.start();
                thread2.start();
        
                Thread.sleep(5000L);
                if(thread1.isAlive()){ 
                    System.out.println("thread1 is alive, thread2 interrupt...");
                    thread2.interrupt();
                    System.out.println("thread2 interrupted...");
                }
            }
        
        }
        /** * thread2 hold lock2... * thread1 hold lock1... * thread1 is alive, thread2 interrupt... * java.lang.InterruptedException * at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireInterruptibly(AbstractQueuedSynchronizer.java:898) * thread2 interrupted... * thread2 release lock2... * thread1 hold lock2... * thread1 release lock1... * thread1 release lock2... */
    
    上面的`thread2.interrupt();`至关重要，如果没有这个中断机制，这两个线程将陷入死锁状态，互相等待对方释放锁，使用`lock1.lockInterruptibly();`表示锁是可以相应线程的中断通知的，一旦线程被中断，这个线程如果正在等待锁，那么这个线程会立马放弃等待锁并中断，但是中断的线程并不会真正完成线程的任务，上面只有thread1完成了所有任务。
 *  申请锁超时时间：上面的中断响应是被动的，需要有中断触发，可不可以在获取锁的时候设置一个超时时间，超过这个时间就不再等待了呢？当然是可以的。
    
        public class ReentrantLockTimeout { 
        
            public static void main(String[] args) { 
                ReentrantLock lock1 = new ReentrantLock();
                ReentrantLock lock2 = new ReentrantLock();
        
                Thread thread1 = new Thread(() -> { 
                    try { 
                        lock1.lockInterruptibly();
                        System.out.println("thread1 hold lock1...");
                        Thread.sleep(1000L);
                        lock2.lockInterruptibly();
                        System.out.println("thread1 hold lock2...");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }finally { 
                        if(lock1.isHeldByCurrentThread()){ 
                            lock1.unlock();
                            System.out.println("thread1 release lock1...");
                        }
                        if(lock2.isHeldByCurrentThread()){ 
                            lock2.unlock();
                            System.out.println("thread1 release lock2...");
                        }
                    }
                });
        
                Thread thread2 = new Thread(() -> { 
                    try { 
                        if(!lock2.tryLock(5, TimeUnit.SECONDS)){ 
                            System.out.println("thread2 wait lock2 5s, give up...");
                        }else{ 
                            System.out.println("thread2 hold lock2...");
                            Thread.sleep(100L);
                        }
                        if(!lock1.tryLock(5, TimeUnit.SECONDS)){ 
                            System.out.println("thread2 wait lock1 5s, give up...");
                        }else{ 
                            System.out.println("thread2 hold lock1...");
                        }
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }finally { 
                        if(lock2.isHeldByCurrentThread()){ 
                            lock2.unlock();
                            System.out.println("thread2 release lock2...");
                        }
                        if(lock1.isHeldByCurrentThread()){ 
                            lock1.unlock();
                            System.out.println("thread2 release lock1...");
                        }
                    }
                });
        
                thread1.start();
                thread2.start();
            }
        
        }
        /** * thread1 hold lock1... * thread2 hold lock2... * thread2 wait lock1 5s, give up... * thread2 release lock2... * thread1 hold lock2... * thread1 release lock1... * thread1 release lock2... */
    
    同样解决了死锁问题，而且更加优雅，不需要中断，但仍然只有一个线程正常完成任务。`tryLock()`方法也可以不带参数，此时线程不会等待，只有锁在不被其它线程占用的时候才会返回`true`，所以可以使用`tryLock()`改进一下上面的程序，不设置超时时间且线程都能全部执行完成。
    
        public class ReentrantLockTryLock { 
        
            public static void main(String[] args) { 
                ReentrantLock lock1 = new ReentrantLock();
                ReentrantLock lock2 = new ReentrantLock();
        
                Thread thread1 = new Thread(() -> { 
                    while (true){ 
                        try { 
                            if(lock1.tryLock()){ 
                                System.out.println("thread1 hold lock1...");
                                Thread.sleep(100L);
                                if(lock2.tryLock()){ 
                                    System.out.println("thread1 hold lock2...");
                                    break;
                                }
                            }
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }finally { 
                            if(lock1.isHeldByCurrentThread()){ 
                                lock1.unlock();
                                System.out.println("thread1 release lock1...");
                            }
                            if(lock2.isHeldByCurrentThread()){ 
                                lock2.unlock();
                                System.out.println("thread1 release lock2...");
                            }
                        }
                    }
                });
        
                Thread thread2 = new Thread(() -> { 
                    while (true){ 
                        try { 
                            if(lock2.tryLock()){ 
                                System.out.println("thread2 hold lock2...");
                                Thread.sleep(100L);
                                if(lock1.tryLock()){ 
                                    System.out.println("thread2 hold lock1...");
                                    break;
                                }
                            }
        
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }finally { 
                            if(lock2.isHeldByCurrentThread()){ 
                                lock2.unlock();
                                System.out.println("thread2 release lock2...");
                            }
                            if(lock1.isHeldByCurrentThread()){ 
                                lock1.unlock();
                                System.out.println("thread2 release lock1...");
                            }
                        }
                    }
                });
        
                thread1.start();
                thread2.start();
            }
        
        }
        /** * thread2 hold lock2... *thread1 hold lock1... *thread2 release lock2... *thread1 release lock1... *thread2 hold lock2... *thread1 hold lock1... *thread1 release lock1... *thread2 hold lock1... *thread2 release lock2... *thread1 hold lock1... *thread2 release lock1... *thread1 hold lock2... *thread1 release lock1... *thread1 release lock2... */
    
    这样其实是一个死循环重复获取和释放锁，总有一次可以同时获取到两个锁。
 *  公平锁：大多数情况下锁是不公平的，即先申请锁的不一定先获得锁，只会在等待队列中随机挑选一个，这就会造成可能有的线程始终得不到执行，就像售票窗口人非常多大家不排队，总有一些挤不进去的人买不到票。而公平锁是按照先来后到挑选的，只要你愿意等，一定能得到执行，重入锁提供了一个构造方法`ReentrantLock(boolean fair)`，当`fair`为`true`时，表示锁是公平的。
    
        public class ReentrantLockFair { 
        
            private static ReentrantLock lock = new ReentrantLock(true);
            //private static ReentrantLock lock = new ReentrantLock();
        
            public static void main(String[] args) throws InterruptedException { 
        
                for (int i = 0; i < 100; i++) { 
                    //确保提交先后顺序
                    Thread.sleep(1L);
                    new PrintThread(i).start();
                }
        
            }
        
            static class PrintThread extends Thread{ 
        
                private int no;
        
                public PrintThread(int no){ 
                    this.no = no;
                }
        
                @Override
                public void run() { 
                    lock.lock();
                    try { 
                        //确保线程占用锁
                        Thread.sleep(1L);
                        System.out.println(String.format("thread%-2s...", no));
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        lock.unlock();
                    }
        
                }
            }
        
        }
    
    上面的程序会一次输出0-99，如果`fair`为`false`或没有`fair`参数，将会看到乱序输出。虽然公平锁好像很完美，但是公平锁会维护一个有序队列，系统开销更大，性能相对也比较低。

重入锁的原子性是由CAS实现的，CAS会在后面总结，重入锁还有挂起和恢复操作(`park()/unpark()`)，这个也会在后面的`LockSupport`总结。

### Condition ###

在并发基础中提到过`Object.wait()`和`Object.notify()`方法，这两个方法用于线程之间通信，并且必须在`synchronized`同步块中调用，如果`ReentrantLock`可以完全替代`synchronized`，那么这个功能也一定要实现。的确如此，`Condition`就是这样的效果，而且比`Object`的两个方法更加灵活，`Condition`有以下基本方法：

*  `await()`：使调用线程进入等待，同时释放当前锁，当其它线程使用`signal()/signalAll()`方法时，线程才有机会获得锁重新执行，可以有超时时间，和`Object.wait()`方法类似；
 *  `awaitUninterruptibly()`：和`await()`方法类似，但是在等待过程中并不会响应中断；
 *  `signal()`：唤醒正在等待的一个线程，`signalAll()`方法则是唤醒所有正在等待同一个`Condition`的线程。

public class ReentrantLockCondition { 
    
        public static void main(String[] args) throws InterruptedException { 
            ReentrantLock lock = new ReentrantLock();
            Condition condition1 = lock.newCondition();
            Condition condition2 = lock.newCondition();
    
            Thread thread1 = new Thread(() -> { 
                lock.lock();
                try { 
                    condition1.await();
                    System.out.println("thread1 end...");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    lock.unlock();
                }
            });
    
            Thread thread2 = new Thread(() -> { 
                lock.lock();
                try { 
                    condition2.await();
                    System.out.println("thread2 end...");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    lock.unlock();
                }
            });
    
            thread1.start();
            thread2.start();
    
            Integer i = new Random().nextInt(1000) % 2;
            System.out.println("i = " + i);
            if(i == 0){ 
                lock.lock();
                condition1.signal();
                lock.unlock();
    
                lock.lock();
                condition2.signal();
                lock.unlock();
            }else{ 
                lock.lock();
                condition2.signal();
                lock.unlock();
    
                lock.lock();
                condition1.signal();
                lock.unlock();
            }
        }
    
    }

上面的程序有两种输出结果，要么是thread1先结束，要么是thread2先结束，这是可以通过`condition.signal()`调用顺序的不同来控制的，而`Object.notify()`并不能实现这种控制，注意不管是`Condition.await()`还是`Object.wait()`，调用这些方法的线程都是让出了锁的，不然别的线程根本没机会调用`notify()`或`signal()`。

重入锁和`Condition`在JDK内部被广泛使用，比如`ArrayBlockingQueue.put()`方法。

### 信号量(Semaphore) ###

无论是`synchronized`还是`ReentrantLock`，一次都之恩能够被一个线程占用，而信号量可以被多个线程同时访问。信号量主要有两个构造函数：

*  `public Semaphore(int permits)`：指定允许同时访问的个数(准入数)，如果一个线程只申请一个信号量，这个permits相当于是制定了同时最多permits个线程执行；
 *  `public Semaphore(int permits, boolean fair)`：同时指定是否是公平的，和重入锁的公平原理一样。

信号量主要有以下方法：

*  `acquire()`：尝试获得一个准入许可，若获取不到会一直等待，知道别的线程释放一个许可或当前线程被中断；
 *  `acquireUninterruptibly()`：和`acquire()`类似，但是不响应中断；
 *  `tryAcquire()`：和`ReeantrantLock.tryLock()`类似，尝试获取一个许可，然后立即返回不会等待，如果获取成功返回`true`，否则返回`false`，当然这个方法也可以设置超时；
 *  `release()`：释放一个许可，这是线程完毕后必须执行的操作。

public class SemaphoreTest { 
    
        public static void main(String[] args) { 
            MyExecutors myExecutors = new MyExecutors(5);
            for (int i = 0; i < 20; i++) { 
                myExecutors.submit(new MyRunnable(i));
            }
            myExecutors.shutdown();
        }
    
        private static class MyExecutors { 
    
            private int n = 3;
    
            private Semaphore semaphore;
    
            List<Thread> threads = new ArrayList<>();
    
            public MyExecutors(int n){ 
                if (n > 3){ 
                    this.n = n;
                }
                semaphore = new Semaphore(n);
            }
    
            public void submit(Runnable runnable){ 
                threads.add(new Thread(() -> { 
                    try { 
                        semaphore.acquire();
                        runnable.run();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }finally { 
                        semaphore.release();
                    }
                }));
            }
    
            public void shutdown(){ 
                if(!threads.isEmpty()){ 
                    for (Thread thread : threads) { 
                        thread.start();
                    }
                }
            }
        }
    
        private static class MyRunnable implements Runnable { 
            private int no;
    
            public MyRunnable(int no){ 
                this.no = no;
            }
    
            @Override
            public void run() { 
                try { 
                    //模拟耗时
                    Thread.sleep(3000L);
                    System.out.println(String.format("i am thread %-2s...", no));
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
    }

上面的代码模拟了一个简易线程池，控制台会分4批，每批5个线程但因出结果，`MyExecutors`把`Runnable`包裹到需要一个信号量准入许可的`Thread`，达到控制同一时间线程池中的允许执行的线程数的效果。

### ReadWriteLock 读写锁 ###

如果系统中有大量的读操作，但是写操作很少的时候，如果读与读之间、读与写之间、写与写都存在竞争，那读写性能会很差，因为读与读之间其实并不会导致数据不一致，只需要读与写、写与写之间保持竞争即可，读的比例越大的系统，使用读写锁的性能提升就越明显。

public class ReadWriteLockTest { 
    
        public static void main(String[] args) throws InterruptedException { 
            reentrantLock();
            reentrantReadWriteLock();
        }
    
        private static void reentrantLock() throws InterruptedException { 
            ReentrantLock reentrantLock = new ReentrantLock();
            Data data = new Data();
            for (int i = 18; i < 20; i++) { 
                int finalI = i;
                new Thread(() -> { 
                    try { 
                        data.setI(reentrantLock, finalI);
                        System.out.println("[" + LocalDateTime.now().toString() + "] reentrant lock write: " + finalI);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }).start();
            }
            for (int i = 0; i < 18; i++) { 
                new Thread(() -> { 
                    try { 
                        int value = data.getI(reentrantLock);
                        System.out.println("[" + LocalDateTime.now().toString() + "] reentrant lock read: " + value);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }).start();
            }
        }
    
        private static void reentrantReadWriteLock() throws InterruptedException { 
            ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
            ReentrantReadWriteLock.ReadLock readLock = reentrantReadWriteLock.readLock();
            ReentrantReadWriteLock.WriteLock writeLock = reentrantReadWriteLock.writeLock();
            Data data = new Data();
            for (int i = 18; i < 20; i++) { 
                int finalI = i;
                new Thread(() -> { 
                    try { 
                        data.setI(writeLock, finalI);
                        System.out.println("[" + LocalDateTime.now().toString() + "] reentrant read write lock write: " + finalI);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }).start();
            }
            for (int i = 0; i < 18; i++) { 
                new Thread(() -> { 
                    try { 
                        int value = data.getI(readLock);
                        System.out.println("[" + LocalDateTime.now().toString() + "] reentrant read write lock read: " + value);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }).start();
            }
        }
    
        private static class Data { 
            private Integer i = 0;
    
            public Integer getI(Lock lock) throws InterruptedException { 
                try { 
                    lock.lock();
                    Thread.sleep(1000L);
                    return i;
                }finally { 
                    lock.unlock();
                }
            }
    
            public void setI(Lock lock, Integer i) throws InterruptedException { 
                try { 
                    lock.lock();
                    Thread.sleep(1000L);
                    this.i = i;
                }finally { 
                    lock.unlock();
                }
            }
        }
    
    }

会发现`ReentrantReadWriteLock`比`ReentrantLock`快多了，因为`ReentrantReadWriteLock`读与读之间是没有竞争的，很适合频繁读很少写的场景。

### CountDownLatch ###

有这样的场景：我需要前面几个线程执行完成以后才开始执行。当然可以使用`join()`来完成，甚至使用`Object.wait()`、`ReeatrantLock.await()`等方法实现，但是这需要知道具体的线程，假设我们现在无法获取对方线程的信息，而且我们只需要知道前面几个线程已经执行完成了就可以了，不需要具体线程信息，那就可以使用`CountDownLatch`。

public class CountDownLatchTest { 
    
        public static void main(String[] args) throws InterruptedException { 
            CountDownLatch countDownLatch = new CountDownLatch(8);
    
            for (int i = 0; i < 10; i++) { 
                int finalI = i;
                new Thread(() -> { 
                    try { 
                        Thread.sleep(1000L);
                        System.out.println(finalI);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }finally { 
                        countDownLatch.countDown();
                    }
                }).start();
            }
    
            countDownLatch.await();
    
            System.out.println("finish...");
        }
    
    }

主线程会在`countDownLatch.awai()`行阻塞，直到`countDownLatch`倒计到0，如果线程完成的数量少于倒计数，主线程将一直等待。

### 循环栅栏(CyclicBarrier) ###

循环栅栏其实跟`CountDownLatch`很相似，唯一的区别是前者可以复用，后者不行；前者是拦截线程数累计到指定数量，后者是递减倒计到0，而且循环栅栏功能更强大，比如可以在栅栏拦截到线程时可以定义一个优先执行的Runnable。

public class CyclicBarrierTest { 
    
        public static void main(String[] args) { 
            CyclicBarrier cyclicBarrier = new CyclicBarrier(10, () -> System.out.println("start..."));
    
            for (int i = 0; i < 20; i++) { 
                int finalI = i;
                new Thread(() -> { 
                    try { 
                        cyclicBarrier.await();
                        System.out.println(finalI);
                        Thread.sleep(1000L);
                    } catch (InterruptedException | BrokenBarrierException e) { 
                        e.printStackTrace();
                    }
                }).start();
            }
        }
    
    }
    /* * start... * 9 * 0 * 2 * 4 * 5 * 6 * 1 * 7 * 3 * 8 * start... * 19 * 11 * 13 * 16 * 10 * 17 * 18 * 15 * 14 * 12 */

上面的例子展示了循环栅栏的复用性，可以一批一批得执行任务。它的实现原理其实也很简单，就是说使用`ReentrantLock`实现的，线程调用到一次`await()`，判断是否需要执行`barrierAction`、是否达到`parties`数量，若达到了就重新初始化`generation`达到可复用的目的。

### 线程阻塞工具类(LockSupport) ###

`LockSupport`是一个非常方便实用的线程阻塞工具，可以使线程在任意位置阻塞。跟`Thread.suspend()`相比，它弥补了如果`resume()`比`suspend()`先调用，线程将无法继续执行的缺点；跟`Object.wait()`相比，它不需要获取任何对象的锁，也不会抛出`InterruptedException`。

`LockSupport`的静态方法`park()`可以阻塞当前线程，当然对应的有`parkNanos()`、`parkUntil()`等超时机制的方法，现在用`LockSupport`重写前面`suspend()`线程挂起导致永久卡死的例子：

public class LockSupportTest { 
    
        public static void main(String[] args) throws InterruptedException { 
            Object object = new Object();
            Thread t0 = new Thread(() -> { 
                synchronized (object){ 
                    System.out.println("[" + LocalDateTime.now().toString() + "] t0 start...");
                    LockSupport.park();
                    System.out.println("[" + LocalDateTime.now().toString() + "] t0 end...");
                }
            });
            Thread t1 = new Thread(() -> { 
                synchronized (object){ 
                    System.out.println("[" + LocalDateTime.now().toString() + "] t1 start...");
                    LockSupport.park();
                    System.out.println("[" + LocalDateTime.now().toString() + "] t1 end...");
                }
            });
    
            t0.start();
            Thread.sleep(100L);
            t1.start();
    
            LockSupport.unpark(t0);
            LockSupport.unpark(t1);
            t0.join();
            t1.join();
        }
    }
    /** * [2018-12-03T21:30:26.454] t0 start... * [2018-12-03T21:30:26.498] t0 end... * [2018-12-03T21:30:26.498] t1 start... * [2018-12-03T21:30:26.498] t1 end... */

上面的例子无论什么情况下都会正常结束，即使`unpark()`调用在`park()`方法之前。因为`LockSupport`机制类似于信号量，它为，每个线程准备了一个许可，如果许可可用，那么`park()`方法就类似与`acquire()`方法，会立即返回，线程继续执行，如果不可用就会阻塞；而`unpark()`方法类似于`release()`方法，另外的线程`unpark()`先于还是后于`park()`都不会影响`park()`获取“许可”。但和信号量不同的是，这里有且只有一个许可。

同时，不同于`suspend()/resume()`的是，如果线程处于阻塞状态，`jstack`打印堆栈信息时，阻塞的线程是一个`WAITING`状态，we且还会显式地说明是由`park()`引起的：

"Thread-0" #12 prio=5 os_prio=0 tid=0x000000001a9d0800 nid=0x4290 waiting on condition [0x000000001b28f000]
       java.lang.Thread.State: WAITING (parking)
            at sun.misc.Unsafe.park(Native Method)
            at java.util.concurrent.locks.LockSupport.park(LockSupport.java:304)
            at github.chenlei.model.LockSupportTest.lambda$main$0(LockSupportTest.java:21)
            - locked <0x00000000d5d1de20> (a java.lang.Object)
            at github.chenlei.model.LockSupportTest$$Lambda$1/1324119927.run(Unknown Source)
            at java.lang.Thread.run(Thread.java:745)

如果`park()`方法改为`park(object)`，还会打印引起阻塞的具体代码行：

"Thread-0" #12 prio=5 os_prio=0 tid=0x000000001ae13800 nid=0x65f4 waiting on condition [0x000000001b6ce000]
       java.lang.Thread.State: WAITING (parking)
            at sun.misc.Unsafe.park(Native Method)
            - parking to wait for  <0x00000000d5d1dd58> (a java.lang.Object)
            at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
            at github.chenlei.model.LockSupportTest.lambda$main$0(LockSupportTest.java:21)
            - locked <0x00000000d5d1dd58> (a java.lang.Object)
            at github.chenlei.model.LockSupportTest$$Lambda$1/1324119927.run(Unknown Source)
            at java.lang.Thread.run(Thread.java:745)

假设线程在`park()`的时候被中断，不会抛出`InterruptedException`，会默默返回，但仍然可以接受中断标志并作出中断响应：

public class LockSupportTest { 
    
        public static void main(String[] args) throws InterruptedException { 
            Object object = new Object();
            Thread t0 = new Thread(() -> { 
                synchronized (object){ 
                    System.out.println("[" + LocalDateTime.now().toString() + "] t0 start...");
                    LockSupport.park(object);
                    if(Thread.interrupted()){ 
                        System.out.println("[" + LocalDateTime.now().toString() + "] t0 interrupted...");
                    }
                    System.out.println("[" + LocalDateTime.now().toString() + "] t0 end...");
                }
            });
            Thread t1 = new Thread(() -> { 
                synchronized (object){ 
                    System.out.println("[" + LocalDateTime.now().toString() + "] t1 start...");
                    LockSupport.park(object);
                    System.out.println("[" + LocalDateTime.now().toString() + "] t1 end...");
                }
            });
    
            t0.start();
            Thread.sleep(100L);
            t1.start();
    
            t0.interrupt();
            LockSupport.unpark(t1);
            t0.join();
            t1.join();
        }
    }
    /** * [2018-12-03T21:49:36.406] t0 start... * [2018-12-03T21:49:36.481] t0 interrupted... * [2018-12-03T21:49:36.481] t0 end... * [2018-12-03T21:49:36.481] t1 start... * [2018-12-03T21:49:36.481] t1 end... */

## 线程池 ##

前面已经用信号量实现了一个简单的线程池。一个普通线程在`run()`方法执行完毕后就会自动被回收，如果下一次需要一个新的线程，就必须new一个新的线程走重复的生命周期。当线程特别多的时候，创建和销毁线程将会占用大量的资源，可能反而得不偿失，而且线程本身也需要空间，大量线程驻留内存将带来巨大的消耗，轻频繁GC，甚至内存溢出。

这时候线程池就有用啦，线程池可以复用线程，同时可以方便控制和管理线程。比如数据库线程池，系统初始化的时候建立几个连接放到线程池；当系统请求变多的时候，再创建新的线程放到线程池中；但线程池中线程不是无限增长的，会有一个上限，当达到这个上限的时候就不再创建新的连接；当一个请求需要一个连接的时候，先从线程池中获取，如果线程池中有可用的连接，就直接返回，如果没有就等待；当一个请求完成时，把连接归还到线程池中供其他请求使用，而不是直接销毁；如果线程池中的线程长时间空闲，超过一定时间后再把这些空闲的线程回收。这样控制了线程的数量，也实现了线程复用，减少了对象频繁的创建和销毁。

### 不要重复造轮子：JDK中的线程池 ###

JDK提供了一套`Executor`框架，它的核心类图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NMX1lE_size_16_color_FFFFFF_t_70]

其中`ThreadPoolExecutor`扮演线程池的角色，`Executors`是一个工厂类，可以通过`Executors`得到不同功能的线程池，如下：

*  `public static ExecutorService newFixedThreadPool(int nThreads)`：返回一个固定数量线程数的线程池，线程池中的数量保持不变。向线程池中提交一个线程，如果有空闲线程，则立即执行；否则放到一个任务队列，待有线程空闲时再提交到线程池执行；
    
        public class FixedThreadPoolTest { 
        
            public static void main(String[] args) { 
                ExecutorService executorService = Executors.newFixedThreadPool(5);
                for (int i = 0; i < 10; i++) { 
                    executorService.submit(() -> { 
                        try { 
                            Thread.sleep(1000L);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                        System.out.println(Thread.currentThread().getName());
                    });
                }
                executorService.shutdown();
            }
        
        }
        /** * pool-1-thread-4 * pool-1-thread-1 * pool-1-thread-5 * pool-1-thread-3 * pool-1-thread-2 * pool-1-thread-4 * pool-1-thread-5 * pool-1-thread-3 * pool-1-thread-2 * pool-1-thread-1 */
    
    可以看出始终是5个线程复用，其中`shutdown()`是关闭线程池，线程池将在所有线程都执行完成后关闭，如果没有关闭操作，线程池将一直可用，程序不会结束；还有一个`shutdownNow()`方法，这个方法会试图中断所有正在执行的线程，但是并不保证一定中断成功，比如那些没有响应中断的线程并不会真正中断，这个方法还会返回一个从未开始执行的线程列表。
 *  `public static ExecutorService newSingleThreadExecutor()`：和上面的一样，不过线程池中线程数只有1个；

public class SingleThreadPoolTest { 
    
        public static void main(String[] args) { 
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            for (int i = 0; i < 10; i++) { 
                executorService.submit(() -> { 
                    try { 
                        Thread.sleep(1000L);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName());
                });
            }
            executorService.shutdown();
        }
    
    }
    /** * pool-1-thread-1 * pool-1-thread-1 * pool-1-thread-1 * pool-1-thread-1 * pool-1-thread-1 * pool-1-thread-1 * pool-1-thread-1 * pool-1-thread-1 * pool-1-thread-1 * pool-1-thread-1 */

可以看出所有任务都在同一个线程中完成。

*  `public static ExecutorService newCachedThreadPool()`：数量不固定的线程池(其实最多只能有232 \- 1线程)，当有新线程提交时，优先使用可复用的空闲线程，否则直接创建新的线程处理。所有线程执行完毕后，返回线程池可复用。
    
        public class CachedThreadPoolTest { 
        
            public static void main(String[] args) throws InterruptedException { 
                ExecutorService executorService = Executors.newCachedThreadPool();
                for (int i = 0; i < 10; i++) { 
                    executorService.submit(() -> { 
                        try { 
                            Thread.sleep(1000L);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                        System.out.println(Thread.currentThread().getName());
                    });
                }
        
                Thread.sleep(2000);
                System.out.println("------------------");
        
                for (int i = 0; i < 10; i++) { 
                    executorService.submit(() -> { 
                        try { 
                            Thread.sleep(1000L);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                        System.out.println(Thread.currentThread().getName());
                    });
                }
                executorService.shutdown();
            }
        
        }
        /** * pool-1-thread-4 * pool-1-thread-5 * pool-1-thread-1 * pool-1-thread-10 * pool-1-thread-2 * pool-1-thread-3 * pool-1-thread-6 * pool-1-thread-7 * pool-1-thread-8 * pool-1-thread-9 * ------------------ * pool-1-thread-2 * pool-1-thread-10 * pool-1-thread-1 * pool-1-thread-3 * pool-1-thread-6 * pool-1-thread-8 * pool-1-thread-5 * pool-1-thread-4 * pool-1-thread-7 * pool-1-thread-9 */
    
    可以看出在来不及复用的时候会直接创建新的线程执行，一旦有线程可以复用，就会使用存在的线程来执行，假设把`Thread.sleep(2000);`替换成`Thread.sleep(60 * 1000);`甚至更长时间，会发第一批执行的线程中现部分线程或所有线程都没得到复用，因为线程池中默认的存活时间是`60s`。
 *  还有两个类似的`SchedulerPool(newScheduledThreadPool() newSingleThreadScheduledExecutor() )`，它们和对应的`ThreadPool`类似，不过`SchedulerExecutorService`扩展了执行计划，能够定是执行任务，比如延时、循环等，可以有三种提交方式，下面以`newScheduledThreadPool()`为例：
    
        public class ScheduledThreadPoolTest { 
        
            public static void main(String[] args) { 
                ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
                System.out.println("[" + LocalDateTime.now().toString() + "] scheduler start...");
                //启动后延迟10秒执行一次
                executorService.schedule(() -> System.out.println("[" + LocalDateTime.now().toString() + "] task delay 10s..."),10, TimeUnit.SECONDS);
                //启动后每5秒执行一次
                executorService.scheduleAtFixedRate(() -> { System.out.println("[" + LocalDateTime.now().toString() + "] task start every 5s...");
                    try { 
                        Thread.sleep(1000L);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                },0, 5, TimeUnit.SECONDS);
                //启动后每次任务执行间隔为5s，即待上次任务结束5s后再执行一次，而不是每5s启动一个新任务，这个任务在同一时间绝对只有一个正在执行
                executorService.scheduleWithFixedDelay(() -> { System.out.println("[" + LocalDateTime.now().toString() + "] task start 5s after last task finished...");
                    try { 
                        Thread.sleep(1000L);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                },0, 5, TimeUnit.SECONDS);
            }
        
        }
        /** * [2018-12-04T21:44:15.231] scheduler start... * [2018-12-04T21:44:15.238] task start every 5s... * [2018-12-04T21:44:15.238] task start 5s after last task finished... * [2018-12-04T21:44:20.238] task start every 5s... * [2018-12-04T21:44:21.241] task start 5s after last task finished... * [2018-12-04T21:44:25.238] task start every 5s... * [2018-12-04T21:44:25.238] task delay 10s... * [2018-12-04T21:44:27.243] task start 5s after last task finished... * [2018-12-04T21:44:30.239] task start every 5s... * [2018-12-04T21:44:33.243] task start 5s after last task finished... * [2018-12-04T21:44:35.238] task start every 5s... * [2018-12-04T21:44:39.245] task start 5s after last task finished... * [2018-12-04T21:44:40.238] task start every 5s... * [2018-12-04T21:44:45.238] task start every 5s... * [2018-12-04T21:44:45.246] task start 5s after last task finished... * [2018-12-04T21:44:50.238] task start every 5s... * [2018-12-04T21:44:51.250] task start 5s after last task finished... * [2018-12-04T21:44:55.239] task start every 5s... * [2018-12-04T21:44:57.251] task start 5s after last task finished... * ... */
    
    这样一下就能看出三种任务计划到底有什么区别啦。

### 线程池实现原理 ###

`Executors`的几个方法中，不管是`newFixedThreadPool()`还是其它几个方法，内部都是用`ThreadPoolExecutor`来实现的，区别在于核心线程数，最大线程数默认都是232 \- 1。  
其中最重要的构造方法是

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

*  `corePoolSize`：核心线程数，指定线程池中常驻线程数，即使这些线程是空闲的也不会被销毁；
 *  `maximumPoolSize`：线程池可接纳最大线程数，决定同一时间线程池中最大活跃线程数；
 *  `keepAliveTime`：如果下次呢哼哧当前线程数超过核心线程数，超出部分的线程处于空闲状态，若在等待`keepAliveTime`时间后仍未等到新的任务，将被销毁；
 *  `unit`：`keepAliveTime`的时间单位；
 *  `workQueue`：任务队列，存储提交但未被执行的任务信息，是一个阻塞队列，可以使用以下集中类型的`BlockingQueue`：
    
     *  直接提交的队列：该功能由`SynchronousQueue`提供，这是一个特殊的队列，没有容量，队列的每一个插入操作都需要等待一个相应的删除操作；反之，每个删除操作都要等待对应的擦如操作。所以这种队列并不会真正保存提交的任务，每提交一个任务都会尝试创建新的线程或复用空闲的线程去执行，如果线程池已满，则执行拒绝策略。因此使用`SynchronousQueue`的线程池通常需要设置很大的`maximumPoolSize`，比如`newCachedThreadPool()`就是用的`SynchronousQueue`；
     *  有界的任务队列：可以使用`ArrayBockingQueue`，构造是需要制定容量`public ArrayBlockingQueue(int capacity)`，队列会有一个最大值，当有新的任务提交时，如果线程池中活跃线程数小于`corePoolSize`则会创建新的线程或复用空闲的线程执行，否则加入等待队列；假设提交任务时等待队列已满，如果线程池活跃线程数小于`maximumPoolSize`，那么会创建新的线程执行，否则执行拒绝策略。可以看出线程池并不能保证任何情况下活跃线程数都不超过`corePoolSize`，当系统繁忙到队列排满时，最大活跃线程数会提高到`maximumPoolSize`，但`newFixedThreadPool()`可以保证，因为它的`corePoolSize = maximumPoolSize`，而且`newFixedThreadPool()`使用的`LinkedBlockingQueue`，看起来是个无解队列，其实也是个最大容量为232 \- 1的队列；
     *  优先任务队列：带有优先级的任务队列，比如`PriorityBlockingQueue`，可以控制队列中任务执行的先后顺序，一班队列都是先进先出算法处理任务的，但是`PriorityBlockingQueue`是按照自然排序或指定`Comparator`升序排列的，即排序后最“小”的任务将优先执行，同时任务必须实现`Comparable`接口且不能为空，虽说是无界队列，其实也有一个上限：232 \- 1 - 8，之所以减8，是一些VM实现数组时会保留一些头信息，减掉了这部分占用的空间。很明显可能会出现饿死的现象，有些排在队列后面的线程可能永远也得不到执行；
 *  `threadFactory`：线程工厂，用于创建线程，做一些统一处理，比如线程组名、线程名、将守护线程设置为非守护线程、设置权限为normal等；
 *  `handler`：拒绝策略，当线程太多处理不过来时，比如等待线程超过了队列的容量等。

任何线程池的调度逻辑可以总结为：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NMX1lE_size_16_color_FFFFFF_t_70 1]  
其中等待执行的任务会在线程池执行完一个任务后从队列中take出来得到执行。

### 拒绝策略 ###

`ThreadPoolExecutor`的最后一个参数定义了拒绝策略，当人物数量超过线程池承载能力时就会走拒绝策略。一般是线程池已满并且队列也已经排满，对于无界队列(伪)不存在拒绝的说法，队列真的被占满的话，会抛出`OutOfMemory`异常，下面模拟一下拒绝的感觉：

public class ThreadPoolReject { 
    
        public static void main(String[] args) { 
            ExecutorService executorService = new ThreadPoolExecutor(
                    5,
                    10,
                    60,
                    TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(10),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.AbortPolicy()
            );
            for (int i = 0; i < 30; i++) { 
                int finalI = i;
                executorService.submit(() -> { 
                    try { 
                        System.out.println(finalI);
                        Thread.sleep(2000L);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
            }
        }
    
    }
    /** * Exception in thread "main" 0 * 2 * 1 * 3 * 4 * 15 * 16 * 17 * 18 * 19 * java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@5fd0d5ae rejected from java.util.concurrent.ThreadPoolExecutor@2d98a335[Running, pool size = 10, active threads = 10, queued tasks = 10, completed tasks = 0] * at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047) * at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823) * at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369) * at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:112) * at github.chenlei.model.ThreadPoolReject.main(ThreadPoolReject.java:27) * 5 * 7 * 9 * 8 * 6 * 10 * 11 * 13 * 12 * 14 */

这个线程池最多可同时有20个活跃的线程，但是一共提交了30个线程，并且每个线程还要休眠2s，导致线程池中存在10个活跃线程，队列中10个线程排满，剩下的10个线程全部拒绝并报`java.util.concurrent.RejectedExecutionException`错。

JDK提供了4中拒绝策略：

*  `AbortPolicy`：抛出异常，超过负载的线程将不会执行，直接被舍弃；
 *  `CallerRunsPolicy`：不抛出异常，线程也不会被舍弃，而是在提交任务的线程中执行或超过负载的线程，从`caller runs`可以看出——我执行不了的，谁提交的谁执行，这可能导致提交任务的线程性能急剧下降；
 *  `DiscardOldestPolicy`：不抛出异常，会有线程被舍弃，舍弃的策略是，当队列排满的时候，尝试挤掉队首的任务，自己添加到队尾，即舍弃队列中最先提交的任务；
 *  `DiscardPolicy`：不跑出异常，超过负载的线程直接舍弃；

如果上面的策略还不能满足需求，可以自己实现`RejectedExecutionHandler`接口，比如我允许系统拒绝任务，但是我想知道多少任务、哪些任务被丢弃了，可以简单实现：

public class ThreadPoolReject { 
    
        public static void main(String[] args) { 
            ExecutorService executorService = new ThreadPoolExecutor(
                    5,
                    10,
                    60,
                    TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(10),
                    Executors.defaultThreadFactory(),
                    (r, executor) -> System.out.println(r.toString() + " is discard...")
            );
            for (int i = 0; i < 30; i++) { 
                int finalI = i;
                executorService.submit(() -> { 
                    try { 
                        System.out.println(finalI);
                        Thread.sleep(2000L);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
            }
        }
    
    }
    /** * 1 * 4 * 0 * 2 * 3 * 15 * 16 * 17 * 18 * 19 * java.util.concurrent.FutureTask@4f3f5b24 is discard... * java.util.concurrent.FutureTask@15aeb7ab is discard... * java.util.concurrent.FutureTask@7b23ec81 is discard... * java.util.concurrent.FutureTask@6acbcfc0 is discard... * java.util.concurrent.FutureTask@5f184fc6 is discard... * java.util.concurrent.FutureTask@3feba861 is discard... * java.util.concurrent.FutureTask@5b480cf9 is discard... * java.util.concurrent.FutureTask@6f496d9f is discard... * java.util.concurrent.FutureTask@723279cf is discard... * java.util.concurrent.FutureTask@10f87f48 is discard... * 5 * 6 * 7 * 10 * 9 * 8 * 11 * 12 * 13 * 14 */

### 自定义ThreadFactory ###

默认的线程工厂做了一些简单的事，比如强制设为非守护线程，为线程分配组、命名、强行公平(线程优先级一样)，如果自己实现一个工厂方法，可以为线程添加很多附加信息、根据业务设置线程优先级，甚至设置所有线程为守护线程，这样当线程池中所有线程执行完毕后，如果主线程还在，那么线程池依然坚挺；一旦主线程退出，线程池会马上退出，不用手动shutdown：

public class DemonThreadFactoryTest { 
    
        public static void main(String[] args) throws InterruptedException { 
            ExecutorService executorService = new ThreadPoolExecutor(
                    5,
                    10,
                    60,
                    TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(10),
                    r -> { 
                        Thread t = new Thread(r);
                        t.setDaemon(true);
                        System.out.println(t.getName() + " created...");
                        return t;
                    },
                    (r, executor) -> System.out.println(r.toString() + " is discard...")
            );
            for (int i = 0; i < 30; i++) { 
                int finalI = i;
                executorService.submit(() -> { 
                    try { 
                        System.out.println(finalI);
                        Thread.sleep(2000L);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
            }
    
            Thread.sleep(5000L);
            executorService.submit(() -> System.out.println("final task..."));
        }
    
    }

上面的程序会在打印`final task...`后直接退出，而不是像普通线程池一样一直处于运行可提交任务状态，因为线程池是否是活的取决于线程池中`Worker`的数量，如果所有`Worker`都是守护线程，主线程结束后，所有`Work`消失，线程池就会执行`tryTerminate()`，如果`Worker`集合是空的，就会终止线程池。

### 线程池扩展 ###

`ThreadPoolExecutor`类是可扩展的，除去核心实现，还在`Worker.runWorker()`方法中提供了一个模板方法：

try { 
        beforeExecute(wt, task);
        Throwable thrown = null;
        try { 
            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();
    }

同时在`tryTerminate()`方法中调用了`terminated()`方法，有助于监控任务的起止和线程池中止的状态。

### submit()与execute() ###

这两个方法都可以提交任务，前者可以提交一个`Future`模式的任务，后者只管执行；假设任务执行异常，前者在不使用`Future`模式的情况下会吞掉一切错误，在使用`Future`时可以在`get()`调用后获取错误，有可能很难定问问题，后者则会抛出错误；前者性能更好，后者要消耗额外的资源。

### 合理选择线程池数量 ###

设一台机器的CPU数量为 n n n，目标CPU的使用率为 u , 0 ≤ u ≤ 1 u, 0 ≤ u ≤ 1 u,0≤u≤1，等待时间与计算时间的比率为 w c \\frac wc cw，则最优线程池大 t h r e a d s threads threads小为：  
 t h r e a d s = n ∗ u ∗ ( 1 + w c ) threads = n \* u \* (1 + \\frac wc) threads=n∗u∗(1\+cw)

### Fork/Join和ForkJoinPool ###

跟`MapReduce`的概念类似：把一个大任务拆分成若干个任务，然后将这些小任务的结果合成，最后得到整个任务的执行结果。这从`Fork/Join`表面意思也能看出来，`fork`是分叉的意思，`join`是合并的意思，在`linux`中`fork()`函数可以创建线程，在`Java`中`join()`表示等待其他线程执行完毕再继续当前线程。

除了`ForkJoin`，还有一个重要概念：工作窃取(`Work-Stealing`)。正常情况下，线程将当前线程任务队列的任务执行完毕后就结束了，但是将实际情况中，即使每个线程都做同样的事情，也可能出现一个线程做完了所有事情，另外一个线程还没有结束，如果完成任务的线程就一直空闲，显然达不到最高执行效率、这时候执行完毕的线程可以去帮助别的未结束的线程执行任务，从为完成线程的任务队列中获取一个任务来执行；需要注意的是，当前线程从队首获取任务，别的线程从队尾获取任务，这样有效避免了竞争。这样的模式成为工作窃取，这也是和`MapReduce`有区别的地方。

下面的例子计算1-n的值

public class ForkJoinTaskTest { 
    
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
    
            System.out.println(100000 + "--------------------");
            compute(100000);
            System.out.println(1000000000 + "--------------------");
            compute(1000000000);
    
        }
    
        private static void compute(int n) throws ExecutionException, InterruptedException { 
            long s = System.currentTimeMillis();
            ForkJoinPool forkJoinPool = new ForkJoinPool();
            ComputeNTask computeNTask = new ComputeNTask(1,n);
            ForkJoinTask<Long> result = forkJoinPool.submit(computeNTask);
            forkJoinPool.shutdown();
    
            System.out.println("fork sum = " + result.get() + "[" + (System.currentTimeMillis() - s) + "]");
    
            s = System.currentTimeMillis();
            long sum = 0;
            for (int i = 1; i <= n; i++) { 
                sum += i;
            }
            System.out.println("for sum = " + sum + "[" + (System.currentTimeMillis() - s) + "]");
        }
    
        static class ComputeNTask extends RecursiveTask<Long> { 
    
            private int start;
    
            private int end;
    
            public ComputeNTask(int start, int end) { 
                this.start = start;
                this.end = end;
            }
    
            @Override
            protected Long compute() { 
                if(start > end){ 
                    throw new IllegalArgumentException();
                }
                long sum = 0L;
                if (end - start < 100000) { 
                    for (int i = start; i <= end; i++) { 
                        sum += i;
                    }
                } else { 
                    List<ComputeNTask> tasks = new ArrayList<>();
                    int step = 99999;
                    int pos = start;
                    while(pos <= end){ 
                        int rEnd = (pos + step <= end) ? (pos + step) : end;
                        ComputeNTask computeNTask = new ComputeNTask(pos, rEnd);
                        tasks.add(computeNTask);
                        computeNTask.fork();
                        pos += step + 1;
                    }
    
                    for (ComputeNTask task : tasks) { 
                        sum += task.join();
                    }
                }
                return sum;
            }
        }
    
    }
    /** * 100000-------------------- * fork sum = 5000050000[5] * for sum = 5000050000[1] * 1000000000-------------------- * fork sum = 500000000500000000[138] * for sum = 500000000500000000[546] */

可以看出，在数字较少的时候，fork/join和for不分伯仲，fork/join没有优势，很大一部分是因为现成的创建、维护等也是要耗时间和资源的；但是当数字变大的时候，差别就非常明显了。

创建一个`ForkJoinTask`可以继承`RecursiveTask<T>`或`RecursiveAction`，前者相当于`Future`模式，后者相当于`Runnable`模式；`Executors`中还有一类线程池`newWorkStealingPool`，返回的就是一个`ForkJoinPool`，可以指定并行数(默认是主机可用的CPU个数)，提交的任务如果是`ForkJoinTask`，会直接走`Fork/Join`模式，否则会用`Adaptor`适配成`ForkJoinTask`，走`Fork/Join`的调用方式而已，并没有真正使用`Fork/Join`模式。

## 并发容器 ##

以前初次尝试多线程的时候，直接使用多线程往`HashMap`添加元素，偶尔会出现卡死的情况，无论是打印堆栈还是断点调试，似乎都找不到为什么。搜索一番发现是`HashMap`并非线程安全的容器，多线程情况下，可能出现多个线程同时往同一个hash地址put元素，因为`HashMap`使用链表存储所有冲突的元素，假设某个地址的尾元素是a，所以可能会出现多个线程同时将a元素的next设置为自己，最后一次设置的线程生出，从而导致元素丢失；甚至是一个线程将a的next设置为b，另外一个线程又将b的next设置为a，形成循环链表，这会导致循环这个链表的时候永远不会结束，导致程序卡死。Java8对这种情况作了优化不会再卡死，但是向同一个位置插入元素导致数量丢失的问题仍然存在，最好实用JDK的并发容器。

*  `ConcurrentHashMap`：首先`Collections.synchronizedMap(Map map)`，这个静态方法会返回一个`SynchronizedMap`，把所有功能都委托给传入的Map实现，自己内部初始化一个`final Object mutex;`负责同步：使用`synchronized(mutex){map.get(key);}`这种委托的方式实现，所有Map相关的操作都必须获得mutex的锁，这会导致高并发场景下并发并不算太高，适用于一般并发场景。更加专业的高并发`HashMap`是`ConcurrentHashMap`，可以理解为线程安全的`HashMap`，Java8中使用`CAS(Compare And Swap)`来实现同步，摒弃了以前的`synchronized + segment`分段锁来实现；
 *  `CopyOnWriteArrayList`：当然也可以使用`Collections.synchronizedList(List list)`实现，实现原理、缺点和上面的类似；`Vector`也是线程安全的List容器，但是使用`synchronized + fail-fast`机制实现，性能较差，`CopyOnWriteArrayList`有更好的性能。`COW(CopyOnWrite)`是一种读写分离的并发控制逻辑，当从容器中读取数据的时候，不需要加锁不需要同步；当向容器中添加数据时，先拷贝原容器到一个新的容器中，然后在这个新的容器里添加元素，最后再把原容器指向新的容器。这样的确可以保证线程安全，但是有两个主要问题：一是占用内存，添加元素时，内存中同时存在两个容器，如果容器本身本来就比较大，那势必会消耗两倍的内存，可能造成频繁的GC，导致系统停顿、相应变慢；二是会出现数据不一致的情况，因为复制容器的时候对写线程是不可见的，可能出现一个线程已经向容器中写入数据，但是另一个线程读不到的情况，如果对一致性要求很高，不建议使用；适用于读操作比例远大于写操作的情况；
 *  `ConcurrentLinkedQueue`：高效并发队列，`CAS`\+用链表实现，可以看作是一个线程安全的`LinkedList`；它的实现因为使用了`CAS`变得异常复杂，但是性能得到了很大的提升；
 *  `BlockingQueue`：这个接口有一系列实现类，通过链表、数组等方式实现，阻塞队列适合作为数据共享通道；其中`Blocking`的意思不是把并发操作变成串行执行的意思，而是在获取和添加元素操作过程上适时`Blocking`。当我们有多个线程同时消费一个队列的时候，怎么知道队列里有新的数据了呢？一种常见的方法是搞个死循环，隔一小段时间取一次，但这样会在队列长时间没有数据的时候造成不必要的资源浪费。`BlockingQueue`有两个入队(`offer()/put()`)和两个出队(`poll()/take()`)的方法，`offer()/poll()`会在入队失败的时候立刻返回`false`，也会在空队列出队时返回`null`，但`put()/take()`方法可能不会立即返回，在队列未满或队列不为空的时候会立即返回，但是队列为空或队列已满的时候，`put()/take()`操作会进入等待，直到队列有元素出队或入队时，向`put()/take()`线程发出信号才会返回，实现原理是`ReentrantLock`;
 *  `ConcurrentSkipListMap`：跳表的实现，是一个Map，跳表相较于`HashMap`在查找性能上要好很多，因为跳表有数据冗余，会存储多个链表，每个链表在不同的层级上，最顶层元素最少，最底层元素最多，从上到下，每一层都是下面所有层的子集，所以最底层就是所有元素；跳表是有序的，遍历跳表会返回一个有序的集合，当需要查找一个元素时，从顶层开始查询，顶层元素最少，一旦命中就返回，如果不能命中就从下层寻找，但是在下层寻找的时候不必从头开始，因为每个层级的链表都是有序的，可以马上确定下一层级中最先从哪个位置开始查询，类似于平衡树，但一个重要的区别是：平衡树的插入和删除可能导致整棵树的调整，但是跳表只需要操作局部数据即可；插入数据的时候会往那一层插入是随机的，因此很可能插入到元素最多的一层，但是即使是最坏的情况也好于从头遍历；同步依然是使用`CAS`实现的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NMX1lE_size_16_color_FFFFFF_t_70]: /images/20220413/e68cfce1c77a4f9f91d461741e7423c0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NMX1lE_size_16_color_FFFFFF_t_70 1]: /images/20220413/2ad1d6dcb9474a10abd25b80d8bbc1b6.png