【Java并发】六、并行模式与算法

Myth丶恋晨 2022-04-04 10:40 241阅读 0赞

# 【Java并发】六、并行模式与算法 #

### 文章目录 ###

*  【Java并发】六、并行模式与算法
 *   *  单利模式
     *  不变模式
     *  生产者-消费者模式
     *  Future模式
     *  并行流水线
     *  并行搜索
     *  Fork/Join与MapReduce
     *  NIO
     *  AIO

## 单利模式 ##

单例模式是设计模式中使用最为普遍的模式之一，它可以确保一些初始化复杂、大对象、核心对象在整个程序中只有一个实例，这不但可以减少内存开销、减轻GC压力(没有频繁的new操作)，也可以保证核心实例的安全性，下面是一个简单的单例模式：

public class SingletonTest { 
    
        public static void main(String[] args) { 
            Singleton instance = Singleton.getInstance();
        }
    
    }
    class Singleton { 
        private static Singleton singleton = new Singleton();
    
        private Singleton(){ 
            System.out.println("singleton created.");
        }
    
        public static Singleton getInstance(){ 
            return singleton;
        }
    }

但是这样的写法有个问题，假设有下面的写法：

public class SingletonTest { 
    
        public static void main(String[] args) { 
            System.out.println(Singleton.STATUS);
        }
    
    }
    class Singleton { 
        private static Singleton singleton = new Singleton();
        
        public static int STATUS = 0;
    
        private Singleton(){ 
            System.out.println("singleton created.");
        }
    
        public static Singleton getInstance(){ 
            return singleton;
        }
    }
    /** * singleton created. * 0 */

`main()`方法中并没有显示调用`getInstance()`方法，但明显在访问`Singleton`的`public static`属性的时候出发了实例的初始化，因为对对象`static`属性的引用会导致对象的所有`static`属性以及`static{}`代码块初始化，但是再不需要初始化的时间初始化，可能并不是我们想要的，比如一些数据其实还没准备好，那么可以改成这样：

public class SingletonTest { 
    
        public static void main(String[] args) { 
            System.out.println(Singleton.STATUS);
        }
    
    }
    class Singleton { 
        private static Singleton singleton = null;
    
        public static int STATUS = 0;
    
        private Singleton(){ 
            System.out.println("singleton created.");
        }
    
        public static synchronized Singleton getInstance(){ 
            if(singleton == null){ 
                singleton = new Singleton();
            }
            return singleton;
        }
    }

并不在定义的时候初始化，而是一开始赋值为`null`，在`getInstance()`方法里判断为空时初始化，非空时直接返回。但是注意到为`getInstance()`放发加上了`synchronized`，因为在并发场景中，可能出现多个线程同时执行初始化，这又导致多项城情况下性能下降，结合以上两种情况可以这样写：

public class SingletonTest { 
    
        public static void main(String[] args) { 
            System.out.println(Singleton.STATUS);
        }
    
    }
    class Singleton { 
    
        public static int STATUS = 0;
    
        private Singleton(){ 
            System.out.println("singleton created.");
        }
    
        private static class SingletonHolder{ 
            private static Singleton singleton = new Singleton();
        }
    
        public synchronized static Singleton getInstance(){ 
            return SingletonHolder.singleton;
        }
    }

这样一样可以控制初始化时间，而且不会影响并发场景下的性能，因为只有在`SingletonHolder`被使用的时候才会导致`singleton`被实例化，只有`getInstance()`才能触发。

## 不变模式 ##

这个还是解决多线程加锁影响性能的问题。假设一个对象创建后就不会再改变，只是作为数据传输使用，不可能交给任何线程修改内部数据，但会被多线程频繁访问，那么这个对象就可以被设计为不变对象。

*  去掉所有`setter()`方法以及导致自身属性修改的方法；
 *  所有属性设置为`private final`，保证不可修改；
 *  确保不能继承或子类无法修改任何属性；
 *  提供构造方法初始化对象。  
    只要满足以上4点，就是一个不变对象，它解决并发线程安全的思想是：我不提供修改的可能，只开放访问的入口，这样就不要同步了。

Java中有很多不变对象，比如所有的包装类以及`String`类。这些类都被广泛地在JDK和应用程序中使用，但我们对这些数据的访问从来都不需要加锁，因为我们知道这些类型是可靠的，一点初始化就没有改变的可能。

## 生产者-消费者模式 ##

生产者-消费者模式是高并发程序中常见的一种设计模式，生产者负责提交任务，消费者负责执行任务，两者之间有一个缓冲队列，用于保存提交的任务供消费者使用。这样设计和好地解决了生产者和消费者之间速度不匹配的问题，增加了系统的吞吐量，比如`kafka`就是这样一种模式，线程池也是类似的，`submit()`任务的线程就是生产者，线程中正在执行的线程就是消费者，存储任务的`BlockingQueue`就是缓冲队列。

但是问题在于：`BlockingQueue`是用加锁阻塞方式实现线程同步的，这样必然会带来一定的性能损耗，如果用CAS实现线程同步的话就非常好了。

好消息是`Disruptor`就是一个基于无锁的队列框架，它是一个环形队列`RingBuffer`，并且初始化的时候必须指定环的大小，所以这不是一个无界队列，但内部还是基于一个普通数组实现的。同时环的大小必须是2的整数次幂，因为这样可以使用`sequence & (queueSize-1)`操作快速将入队的元素放到数组对应的位置上，这比取余快多了。因为环的大小是固定的，所以不会出现空间分配和回收，可以做到内存复用，减少分配和回收的系统开销，下面直接给出`Disruptor`和`ArrayBolckingQueue`的性能差距测试。  
先定义一个`Event`包含需要处理的数据

public class LongEvent { 
        private long value;
    
        public void set(long value) { 
            this.value = value;
        }
    }

先是`ArrayBlockingQueue`实现:

public class ArrayBlockingQueueTest { 
    
        private static ArrayBlockingQueue<LongEvent> queue = new ArrayBlockingQueue<>(1024 * 32);
    
        public static void main(String[] args) { 
            ExecutorService service = Executors.newCachedThreadPool(DaemonThreadFactory.INSTANCE);
    
            class Consumer implements Runnable { 
                @Override
                public void run() { 
                    try { 
                        while (true){ 
                            LongEvent data = queue.take();
                        }
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
    
            service.submit(new Consumer());//13234
    
            ByteBuffer bb = ByteBuffer.allocate(8);
            long start = System.currentTimeMillis();
            for (long l = 0; l < 1024 * 1024 * 100; l++) { 
                bb.putLong(0, l);
                LongEvent event = new LongEvent();
                event.set(bb.getLong(0));
                queue.offer(event);
            }
    
            Runtime.getRuntime().addShutdownHook(new Thread(() -> System.out.println(System.currentTimeMillis() - start)));
        }
    }

然后是`Disruptor`实现：

public class DisruptorTest { 
    
        public static void handleEvent(LongEvent event, long sequence, boolean endOfBatch) { 
            //System.out.println(event);
        }
        public static void handleEventsWithWorkerPool(LongEvent event) { 
            //System.out.println(event);
        }
    
        public static void translate(LongEvent event, long sequence, ByteBuffer buffer) { 
            event.set(buffer.getLong(0));
        }
    
        public static void main(String[] args) throws Exception { 
            // Specify the size of the ring buffer, must be power of 2.
            int bufferSize = 1024 * 32;
    
            // Construct the Disruptor
            Disruptor<LongEvent> disruptor = new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
    
            // Connect the handler
            disruptor.handleEventsWith(Main::handleEvent);//9603
    
            // Start the Disruptor, starts all threads running
            disruptor.start();
    
            // Get the ring buffer from the Disruptor to be used for publishing.
            RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
    
            ByteBuffer bb = ByteBuffer.allocate(8);
            long start = System.currentTimeMillis();
            for (long l = 0; l < 1024 * 1024 * 100; l++) { 
                bb.putLong(0, l);
                ringBuffer.publishEvent(DisruptorTest::translate, bb);
                //Thread.sleep(1000);
            }
    
            Runtime.getRuntime().addShutdownHook(new Thread(() -> System.out.println(System.currentTimeMillis() - start)));
        }
    
    }

发现差别在3秒左右，这是在单个`consumer`的情况下。  
两个`consumer`的情况下(我的电脑是双核)

public class ArrayBlockingQueueTest { 
    
        private static ArrayBlockingQueue<LongEvent> queue = new ArrayBlockingQueue<>(1024 * 32);
    
        public static void main(String[] args) { 
            ExecutorService service = Executors.newCachedThreadPool(DaemonThreadFactory.INSTANCE);
    
            class Consumer implements Runnable { 
                @Override
                public void run() { 
                    try { 
                        while (true){ 
                            LongEvent data = queue.take();
                        }
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
    
            service.submit(new Consumer());
            service.submit(new Consumer());//two consumers 6514
    
            ByteBuffer bb = ByteBuffer.allocate(8);
            long start = System.currentTimeMillis();
            for (long l = 0; l < 1024 * 1024 * 100; l++) { 
                bb.putLong(0, l);
                LongEvent event = new LongEvent();
                event.set(bb.getLong(0));
                queue.offer(event);
            }
    
            Runtime.getRuntime().addShutdownHook(new Thread(() -> System.out.println(System.currentTimeMillis() - start)));
        }
    
    }

public class DisruptorTest { 
    
        public static void handleEvent(LongEvent event, long sequence, boolean endOfBatch) { 
            //System.out.println(event);
        }
        public static void handleEventsWithWorkerPool(LongEvent event) { 
            //System.out.println(event);
        }
    
        public static void translate(LongEvent event, long sequence, ByteBuffer buffer) { 
            event.set(buffer.getLong(0));
        }
    
        public static void main(String[] args) throws Exception { 
            // Specify the size of the ring buffer, must be power of 2.
            int bufferSize = 1024 * 32;
    
            // Construct the Disruptor
            Disruptor<LongEvent> disruptor = new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE, ProducerType.MULTI, new BusySpinWaitStrategy());
    
            // Connect the handler
            //disruptor.handleEventsWith(DisruptorTest::handleEvent);
    
            disruptor.handleEventsWithWorkerPool(
                    DisruptorTest::handleEventsWithWorkerPool,
                    DisruptorTest::handleEventsWithWorkerPool
            );//two consumers 4998
    
            // Start the Disruptor, starts all threads running
            disruptor.start();
    
            // Get the ring buffer from the Disruptor to be used for publishing.
            RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
    
            ByteBuffer bb = ByteBuffer.allocate(8);
            long start = System.currentTimeMillis();
            for (long l = 0; l < 1024 * 1024 * 100; l++) { 
                bb.putLong(0, l);
                ringBuffer.publishEvent(DisruptorTest::translate, bb);
                //Thread.sleep(1000);
            }
    
            Runtime.getRuntime().addShutdownHook(new Thread(() -> System.out.println(System.currentTimeMillis() - start)));
        }
    
    }

相差在2s左右，当CPU核数越多的时候，两者的差距会越来越大。  
更多`Disruptor`用法：  
[Disruptor系列3：Disruptor样例实战][Disruptor_3_Disruptor]  
[disruptor][]

我们知道`BlockingQueue`通知消费者有新数据产生的方式是：消费者不停轮询，如果队列中有数据就直接返回，否则就挂起，直到某个入队操作唤醒，再从队列中返回一个数据。`Disruptor`有多种通知(等待)策略：

*  `BlockingWaitStrategy`：默认策略，和`BlockingQueue`的实现方式如初一则，都需要使用`ReentrantLock/Condition`，节省CPU，但是高并发情况下表现最差；
 *  `SleepingWaitStrategy`：这个策略也是基于对CPU的保守使用，虽然是在循环中不断等待数据，但是它会自旋等待一段时间，如果不成功则使用`Thread.yield()`让出CPU，最终使用`LockSupport.parkNanos(1)`进行线程休眠，以确保不占用太多的CPU。这个策略会有较高的平均延时，但是对其它线程的影响较小，适合用于实时性要求不是特别高而且系统中还有其他核心业务的场景；
 *  `YieldingWaitStrategy`：同样是不断循环，也会使用`Thread.yield()`让出CPU，但是少了自旋和休眠，会有更高的效率，但是对CPU就没有上面的写略那么友好了，消费线程只是一个执行了`Thread.yield()`的死循环，所以消费线程最好低于系统的逻辑CPU数量（4核8线程的8），否则整个应用程序都可能受到影响——除非这个系统只干消费这么一件事。这种策略适合于低延迟但又为其他线程保留可执行余地的场景；
 *  `BusySpinWaitStrategy`：这是最疯狂的一种策略，就是一个死循环！消费线程会尽最大努力疯狂从队列中获取数据，因此它会毫不犹豫地吃掉所有CPU资源，所以消费线程数一定得小于物理CPU数量（4核8线程的4）。这种策略用于真的非常繁忙或对实时性要求较高的场景。

除了CAS、等待策略等会影响整个队列的吞吐量以外，还有一个可能会影响性能的点就是伪共享。为了提高CPU速度，CPU有一个高速缓存Cache，每次读取一个变量时，会把邻近的几个变量一起拉到高速缓存中（缓存行，一般为64字节），因为CPU访问高速缓存的速度非常快。但是这个缓存优化却是一个潜在的性能杀手：假设有两个CPU C1和C2，有两个队列中的数据A、B，假设A/B同在一个缓存行且同时被C1和C2拉到各自的高速缓存，这时C1更新了A，C1为了保障修改的值被C2看到，基于内存屏障的机制，会将修改的变量立即刷新到主内存中；这时候C2中的缓存行因为包含A，会导致整个C2的缓存行失效，即使C2只关心B也只能到主存中取值。如果CPU经常不能命中缓存，队列的吞吐量就会急剧下降，这就是伪共享。

那么Disruptor是怎么解决伪共享的问题的呢？就是通过缓存行填充。既然每个CPU会将访问的变量相邻的64字节的变量拉倒自己的内存空间，那么可以在该变量上再新建几个空变量满足64个字节不就可以了么。即使出现伪共享，也不会影响其他CPU。所以在`Disruptor`的`RingBuffer`源码中可以看到有几个Long类型的变量`P1,P2,P3,P4,P5,P6,P7`，就是为了填充，比如频繁使用的`sequence`变量。

public class VolatileLongTest { 
        private final static int CORES = 4;
        private final static long N_THREADS = (long) 500 * 1000 * 1000;
    
    
        public static void main(String[] args) { 
            testVolatileLong();
        }
    
        private static void testVolatileLong(){ 
            VolatileLong[] longs = new VolatileLong[CORES];
            for (int i = 0; i < longs.length; i++) { 
                longs[i] = new VolatileLong();
            }
            long s = System.currentTimeMillis();
            for (int i = 0; i < CORES; i++) { 
                int finalI = i;
                new Thread(() -> { 
                    for (long l = 0; l < N_THREADS; l++) { 
                        longs[finalI].value = l;
                    }
                }).start();
            }
            Runtime.getRuntime().addShutdownHook(new Thread(() -> System.out.println((System.currentTimeMillis() - s))));
        }
    
        public static class VolatileLong { 
            public volatile long value = 0;
            public long p1=0L,p2=0L,p3=0L,p4=0L,p5=0L,p6=0L,p7=0L;//false cache
        }
        /** * 注释掉false cache行：15864 * 不注释false cache行：4210 */
    
    }

从上面的例子看来，伪共享对性能的影响还是肉眼可见的，对于类似数组这种连续地址的频繁更新操作，伪共享优化还是有必要的。

## Future模式 ##

`Future`模式是很常见的一种并发设计模式，它的核心思想是异步调用，它不需要线程立即返回真实结果，只返回一个契约，调用者线程依然可以毫无阻碍地做其他事情，等到将来需要异步线程的结果的时候再根据先前的契约获取结果，获取结果的过程是阻塞的。

`Future`相比于`Thread.join()`，后者会导致等待线程结果的这段时间，调用者线程无法做任何事情，即使知道异步线程需要耗费相当长的时间；而前者会立即返回，线程可以继续做其他事情，即使做其他事情的过程中所有异步线程就已经结束也无需担心，任何时候都可以凭借"契约"得到结果。

Java中有一套实现好的`Future`模式，使用起来也非常方便：

public class FutureTaskTest { 
    
        public static void main(String[] args) throws InterruptedException, ExecutionException { 
            FutureTask future = new FutureTask<>(() -> { 
                System.out.println("[" + LocalDateTime.now().toString() + "] future start...");
                Thread.sleep(10000L);
                System.out.println("[" + LocalDateTime.now().toString() + "] future end...");
                return "finish";
            });
            new Thread(future).start();
    
            System.out.println("[" + LocalDateTime.now().toString() + "] do something start...");
            Thread.sleep(3000);
            System.out.println("[" + LocalDateTime.now().toString() + "] do something end...");
    
            System.out.println("[" + LocalDateTime.now().toString() + "] do another thing start...");
            Thread.sleep(5000);
            System.out.println("[" + LocalDateTime.now().toString() + "] do another thing end...");
    
            System.out.println("[" + LocalDateTime.now().toString() + "] future result: " + future.get());
        }
    
    }
    /** * [2018-12-19T21:03:49.960] future start... * [2018-12-19T21:03:49.960] do something start... * [2018-12-19T21:03:52.960] do something end... * [2018-12-19T21:03:52.960] do another thing start... * [2018-12-19T21:03:57.961] do another thing end... * [2018-12-19T21:03:59.961] future end... * [2018-12-19T21:03:57.961] future result: finish */

一般情况下`Future`不这么使用，都是配合`ExecutorService`线程池来使用的，线程池的`submit()`方法都会返回一个`Future`，每次提交任务到线程池中就可以得到一个"契约"，可以根据这个契约判断任务是否完成或者获取结果。

public class ExecutorFutureTest { 
    
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
            ExecutorService executorService = Executors.newFixedThreadPool(3);
            Future futureRunnable = executorService.submit(() -> { 
                try { 
                    Thread.sleep(3000L);
                    System.out.println("runnable");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            });
            Future<String> futureRunnableT = executorService.submit(() -> { 
                try { 
                    Thread.sleep(2000L);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }, "runnableT");
            Future<String> futureCallable = executorService.submit(() -> { 
                Thread.sleep(1000L);
                return "callable";
            });
            executorService.shutdown();
    
            Thread.sleep(1000L);
    
            System.out.println("futureRunnable: " + futureRunnable.isDone());
            System.out.println("futureRunnableT: " + futureRunnableT.isDone());
            System.out.println("futureCallable: " + futureCallable.isDone());
    
            System.out.println("futureRunnable: " + futureRunnable.get());
            System.out.println("futureRunnableT: " + futureRunnableT.get());
            System.out.println("futureCallable: " + futureCallable.get());
        }
    
    }
    /** * futureRunnable: false * futureRunnableT: false * futureCallable: true * runnable * futureRunnable: null * futureRunnableT: runnableT * futureCallable: callable */

除了普通的`Future`，还有一类特殊的`Future: CompletableFuture`。普通的`Future`还是有一定的局限性的，如果说`Future`让调用者获取线程结果的操作变得更加主动和可控，那么它也应该提供线程之间相互依赖时的解决方案，比如一个`Future`依赖于另一个`Future`的结果，但不幸的是并不能直接实现这样的操作，但如果依然使用普通线程的等待、阻塞、`join()`等操作，那`Future`的优势荡然无存，`CompletableFuture`就是线程之间相互依赖的`Future`解决方案。

`CompletableFuture`是我再看Java8心特性的时候发现的，具体的骚操作移步：[CompletableFuture 详解][CompletableFuture]，这里有详细的讲解，几乎可以满足绝大部分线程依赖的场景，而且同时提供了阻塞和非阻塞的方法。

## 并行流水线 ##

假设一个任务有ABCD四个步骤，并且每个步骤都依赖于前一个步骤的结果，这样的任务是没有必要并行的，不如在一个任务里一次执行ABCD四个步骤就行了。

但是任务数量很多的时候情况就有些不一样了：如果使用一个任务一个任务地执行肯定会很慢，但是如果用一批线程来做，那会让很多任务一直没有开始，最后任务完成也是分散不连续的，可能会影响吞吐量；把所有步骤写在一个线程也不是很好的办法，当任务失败需要重试的时候甚至不知道从哪一步开始重试，流水线可以解决这两个问题，它既可以达到多线程的处理速度，也可以增加系统的吞吐量，甚至可以记录各个步骤的状态，实现重试时忽略已经完成的步骤：

public class StreamTest { 
    
        public static void main(String[] args) throws InterruptedException { 
            List<Integer> list = new ArrayList<>();
            for (int i = 0; i < 10000; i++) { 
                list.add(i);
            }
    
            foreach(list);//52299
            //executorService(list);//13061
            //stream(list);//13185
        }
    
        private static void executorService(List<Integer> list) { 
            long s = System.currentTimeMillis();
    
            ExecutorService executorService = Executors.newFixedThreadPool(4);
            for (Integer integer : list) { 
                executorService.execute(() -> { 
                    try { 
                        Thread.sleep(1);
                        Thread.sleep(1);
                        Thread.sleep(1);
                        Thread.sleep(1);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
            }
            executorService.shutdown();
    
            Runtime.getRuntime().addShutdownHook(new Thread(() -> System.out.println(System.currentTimeMillis() - s)));
        }
    
        private static void stream(List<Integer> list) throws InterruptedException { 
            long s = System.currentTimeMillis();
            ArrayBlockingQueue<Integer> a = new ArrayBlockingQueue<>(1000);
            ArrayBlockingQueue<Integer> b = new ArrayBlockingQueue<>(1000);
            ArrayBlockingQueue<Integer> c = new ArrayBlockingQueue<>(1000);
            ArrayBlockingQueue<Integer> d = new ArrayBlockingQueue<>(1000);
            new Thread(() -> { 
                try { 
                    while (true) { 
                        Integer integer = a.take();
                        Thread.sleep(1L);
                        b.put(integer);
                    }
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }).start();
            new Thread(() -> { 
                try { 
                    while (true) { 
                        Integer integer = b.take();
                        Thread.sleep(1L);
                        c.put(integer);
                    }
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }).start();
            new Thread(() -> { 
                try { 
                    while (true) { 
                        Integer integer = c.take();
                        Thread.sleep(1L);
                        d.put(integer);
                    }
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }).start();
            new Thread(() -> { 
                try { 
                    while (true) { 
                        Integer integer = d.take();
                        Thread.sleep(1L);
                        if (integer == 9999) { 
                            System.out.println(System.currentTimeMillis() - s);
                            System.exit(0);
                        }
                    }
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }).start();
    
            for (Integer integer : list) { 
                a.put(integer);
            }
        }
    
        private static void foreach(List<Integer> list) throws InterruptedException { 
            long s = System.currentTimeMillis();
    
            for (Integer integer : list) { 
                Thread.sleep(1L);//A
                Thread.sleep(1L);//B
                Thread.sleep(1L);//C
                Thread.sleep(1L);//D
            }
    
            Runtime.getRuntime().addShutdownHook(new Thread(() -> System.out.println(System.currentTimeMillis() - s)));
        }
    
    }

可以看出多线程和流水线耗时是差不多的，如果多线程中有两个任务的某个步骤卡住了，就会长时间丢失两个可用线程，即使这两个线程可以在等待的时间里做其他事，流水线即使有某些步骤卡住了，其他步骤还是会正常执行；而且流水线因为每个步骤都有缓存，每个步骤又可以用多个线程来消耗，所以可以增加系统的吞吐量。

## 并行搜索 ##

并行搜索意味着各个搜索线程之间会有通信机制，因为一旦某一个线程搜索到结果，其它线程就可以结束了。并行搜索的实现也并不困难：将数据集分成几部分，然后各部分单独搜索即可，一旦某个线程搜索到结果就通知其它线程结束并返回这个搜索结果。

最值得思考的部分是如何进行线程之间的通信，最简单的方法是设置一个公共变量，如果某个线程搜索到了结果就更改这个变量(`volatile` \+ `synchronized`)，其它线程发现更新就主动中断当前线程。

## Fork/Join与MapReduce ##

前面JDK并发包中提到过`Fork/Join`框架和`ForkJoinPool`，采用递归的方式定义任务拆分和合并的规则，不断的使用分治的思想，到一定规模后启动进程计算，`Fork/Join`的独特之处在于工作窃取(`Work-Stealing`)：当`Fork/Join`模式下的某个线程执行完毕处于空闲状态，它可以到别的线程的任务队列队尾获取任务执行，使得多核、多线程的优势发挥到最大。

`Hadoop MapReduce`的核心思想跟`Fork/Join`是一样的，都是分而治之。`MapReduce`分`map`阶段和`reduce`阶段，每个阶段都是用键值对（`key/value`）作为输入和输出。而程序员要做的就是定义好这两个阶段的函数：`map`函数和`reduce`函数。客户端配置好`map`和`reduce`，以`job`的形式提交给`JobTracker`；接着`JobTracker`将需要的文件复制到`HDFS`，包括`MapReduce`程序打包的`JAR`文件、配置文件和客户端计算所得的输入划分信息，然后将任务放到任务队列；任务调度器根据自己的算法调度到改作业，为每个划分创建一个`map`并提交到`TaskTracker`，`TaskTracker`根据主机核的数量和内存的大小有固定数量的`map`槽和`reduce`槽，其中`map`任务会分配给对应主机上有目标数据块的`TaskTracker`，同时把`jar`包复制到该机器上(数据本地化，运算移动、数据不移动)；`TaskTracker`每隔一段时间会给`JobTracker`发送一个心跳，告诉`JobTracker`它依然在运行，同时心跳中还携带着很多的信息，比如当前`map`任务完成的进度等信息。当`JobTracker`收到作业的最后一个任务完成信息时，便把该作业设置成“成功”；最后进入`reduce`阶段，拷贝不同分区的数据到不同的`TaskTracker`节点，执行完毕后写入`HDFS output`。

可以看出`Fork/Join`更像是单机版的`MapReduce`，增加了工作窃取，同时`MapReduce`的任务状态监听变成了多线程之间的通信和控制；`MapReduce`像是分布式的`Fork/Join`，没有工作窃取。`MapReduce`和`Fork/Join`的执行时间都取决于执行时间最长的节点/线程，但是`Fork/Join`为了让这种影响减少到最小，使用了工作窃取算法来尽量加快最慢线程的执行。

## NIO ##

`NIO`是`New IO`的简称，是一套可以完全替代Java IO的新的IO机制。严格来说，NIO与并发其实没有直接联系，但是用NIO可以大大提高并发处理效率。而并发较高的场景一般是Socket网络读写，因此这里基本只讨论网络IO。

我们先看看一个传统的Socket-Server-Client：  
一个简单的echo Server：

public class Server { 
    
        private final static ExecutorService POOL = Executors.newCachedThreadPool();
    
        static class Handler implements Runnable { 
    
            Socket socket;
    
            public Handler(Socket socket) { 
                this.socket = socket;
            }
    
            @Override
            public void run() { 
                BufferedReader br = null;
                PrintWriter pr = null;
    
                try { 
    
                    br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    pr = new PrintWriter(socket.getOutputStream(), true);
    
                    String line;
                    long s = System.currentTimeMillis();
                    while((line = br.readLine()) != null){ 
                        pr.write(line);
                    }
                    System.out.println(Thread.currentThread().getName() + ": " + (System.currentTimeMillis() - s) + "ms");
    
                } catch (IOException e) { 
                    e.printStackTrace();
                } finally { 
                    try { 
                        if (br != null) br.close();
                        if (pr != null) br.close();
                        socket.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    
        public static void main(String[] args) { 
            ServerSocket serverSocket;
            Socket socket;
            try { 
                serverSocket = new ServerSocket(8000);
                while(true){ 
                    socket = serverSocket.accept();
                    System.out.println(socket.getRemoteSocketAddress() + ": connect!");
                    POOL.execute(new Handler(socket));
                }
            }catch (IOException e){ 
                e.printStackTrace();
            }
    
        }
    
    }

模拟高并发的客户端Client：

public class Client { 
    
        private final static ExecutorService POOL = Executors.newFixedThreadPool(10);
        private final static int MILLS = 1000;
    
        static class ClientRunnable implements Runnable { 
            @Override
            public void run() { 
                Socket client = null;
                PrintWriter pr = null;
                BufferedReader br = null;
                try { 
                    client = new Socket();
                    client.connect(new InetSocketAddress("localhost", 8000));
                    pr = new PrintWriter(client.getOutputStream(), true);
                    pr.print("H");
                    Thread.sleep(MILLS);
                    pr.print("e");
                    Thread.sleep(MILLS);
                    pr.print("l");
                    Thread.sleep(MILLS);
                    pr.print("l");
                    Thread.sleep(MILLS);
                    pr.print("o");
                    Thread.sleep(MILLS);
                    pr.print("!");
                    Thread.sleep(MILLS);
                    pr.println();
                    pr.flush();
    
                    br = new BufferedReader(new InputStreamReader(client.getInputStream()));
                    System.out.println("server response: " + br.readLine());
                } catch (Exception e) { 
                    e.printStackTrace();
                } finally { 
                    try { 
                        if(br != null) br.close();
                        if(pr != null) pr.close();
                        if(client != null) client.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    
        public static void main(String[] args) { 
            for (int i = 0; i < 10; i++) { 
                POOL.execute(new ClientRunnable());
            }
        }
    
    }

先启动Server再启动Client，最后Server输出如下：

/127.0.0.1:55473: connect!
    /127.0.0.1:55472: connect!
    /127.0.0.1:55474: connect!
    /127.0.0.1:55475: connect!
    /127.0.0.1:55478: connect!
    /127.0.0.1:55476: connect!
    /127.0.0.1:55477: connect!
    /127.0.0.1:55471: connect!
    /127.0.0.1:55479: connect!
    /127.0.0.1:55480: connect!
    pool-1-thread-2: 6006ms
    pool-1-thread-1: 6006ms
    pool-1-thread-7: 6006ms
    pool-1-thread-6: 6005ms
    pool-1-thread-8: 6007ms
    pool-1-thread-5: 6006ms
    pool-1-thread-10: 6007ms
    pool-1-thread-4: 6009ms
    pool-1-thread-3: 6008ms
    pool-1-thread-9: 6008ms

发现服端每个请求几乎要耗时6s处理，假设并发的规模足够大，每个请求速度又不够快，那么服务器的资源将会很快被耗尽，或者能够处理的并发数大幅减少，然而导致服务器处理性能下降得并不是服务器本身资源或性能问题，而是CPU一直在等待网络IO，说得更直白一点，就是一直处于空闲但是又无能为力。

在等待IO的这段时间内，完全可以把CPU让出来给其他真正需要IO的线程使用，如何将等待IO的时间分离出来就是NIO可以做到的事。NIO的几个重要概念，这里用快递业务来做类比：

*  `Channel`：通道，一个`Channel`对应一个`Socket`，如果把客户端和服务端的`Socket`比作收发货地址，那`Channel`可以看做发货地址与收货地址间的固定线路；
 *  `Buffer`：是一个`byte`数组，所有发往`Chanel`的数据必须打包成一个`Buffer`，所以`Buffer`可以看作一个打包好的包裹，这个包裹可以从收货地址发出，也可以从发货地址发出；
 *  `Selector`：现在客户端和服务端有了通道、定义好了数据交互方式，但是彼此不知道对方什么时候发送的数据，也不知道什么时候接收数据，这时候就需要一个`Selector`来管理通道，`Channel`先到`Selector`中注册，当某个`Channel`有数据准备好时，`Selector`就会接到通知，得到那些数据，通过合适的方式处理并传输这些数据，所以`Selector`可以看作是一个快递员，他负责多个路线的快递的收件和派送，收寄双方不需要直接通知对方，只需要把包裹交给快递员就好了，快递员完成一系列的内部处理(数据处理)后，再把处理后的数据送往目的地。

这样的好处是多个请求可以由一个或极少数的`Selector`管理，当`Channel`中有数据准备好的时候，`Selector`开始工作，否则处于等待状态，但是因为`Selector`是极少数的，不必担心占用太多的服务器资源。我们现在用NIO来构造一个Socket-Server-Client，Server：

public class NioServer { 
    
        private Selector selector;
        private ExecutorService POOL = Executors.newCachedThreadPool();
        private final static Map<Socket,Long> TIME_STAT_MAP = new HashMap<>();
    
        private void startServer() throws Exception { 
            selector = SelectorProvider.provider().openSelector();
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.configureBlocking(false);
    
            InetSocketAddress isa = new InetSocketAddress(8000);
            ssc.socket().bind(isa);
    
            /*SelectionKey acceptKey = */ssc.register(selector, SelectionKey.OP_ACCEPT);
    
            while (true) { 
                selector.select();
                Set readKeys = selector.selectedKeys();
                Iterator iterator = readKeys.iterator();
                while (iterator.hasNext()){ 
                    SelectionKey key = (SelectionKey) iterator.next();
                    iterator.remove();
    
                    if(key.isAcceptable()){ 
                        doAccept(key);
                    }else if(key.isValid() && key.isReadable()){ 
                        TIME_STAT_MAP.put(((SocketChannel)key.channel()).socket(),System.currentTimeMillis());
                        doRead(key);
                    }else if(key.isValid() && key.isWritable()){ 
                        doWrite(key);
                        Socket socket = ((SocketChannel)key.channel()).socket();
                        System.out.println("socket: " + socket + ", spend: " + (System.currentTimeMillis() - TIME_STAT_MAP.get(socket)) + "ms");
                    }
                }
            }
        }
    
        private void doWrite(SelectionKey key) { 
            SocketChannel channel = (SocketChannel) key.channel();
            LinkedList<ByteBuffer> byteBuffers = (LinkedList<ByteBuffer>) key.attachment();
    
            ByteBuffer bb = byteBuffers.getLast();
            try { 
                int len = channel.write(bb);
                if(len == -1){ 
                    disconnect(key);
                    return;
                }
    
                if(bb.remaining() == 0){ 
                    byteBuffers.removeLast();
                }
    
                if(byteBuffers.size() == 0){ 
                    key.interestOps(SelectionKey.OP_READ);
                }
    
            }catch (Exception e){ 
                e.printStackTrace();
                disconnect(key);
            }
        }
    
        private void disconnect(SelectionKey key) { 
            try { 
                key.channel().close();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    
        private void doRead(SelectionKey key) { 
            SocketChannel channel = (SocketChannel) key.channel();
            ByteBuffer bb = ByteBuffer.allocate(8192);
    
            try{ 
                int len = channel.read(bb);
                if(len < 0){ 
                    disconnect(key);
                    return;
                }
            }catch (Exception e){ 
                e.printStackTrace();
                disconnect(key);
            }
    
            bb.flip();
            POOL.execute(new Handler(key, bb));
        }
    
        private void doAccept(SelectionKey key) { 
    
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            SocketChannel client;
            try{ 
                client = server.accept();
                client.configureBlocking(false);
    
                //告诉快递员快递打包好了，快来收件
                SelectionKey clientKey = client.register(selector, SelectionKey.OP_READ);
                clientKey.attach(new LinkedList<ByteBuffer>());
            }catch (Exception e){ 
                e.printStackTrace();
            }
        }
    
        private class Handler implements Runnable { 
            SelectionKey key;
            ByteBuffer bb;
    
            public Handler(SelectionKey key, ByteBuffer bb) { 
                this.key = key;
                this.bb = bb;
            }
    
            @Override
            public void run() { 
                ((LinkedList<ByteBuffer>) key.attachment()).add(bb);
                key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
                selector.wakeup();
            }
        }
    
        public static void main(String[] args) throws Exception { 
            NioServer server = new NioServer();
            server.startServer();
        }
    }

Client：

public class NioClient { 
    
        private final static ExecutorService POOL = Executors.newFixedThreadPool(10);
    
        public static void main(String args[]) throws Exception { 
    
            for (int i = 0; i < 10; i++) { 
                POOL.execute(new NioSocketRunnable());
            }
    
        }
    
        static class NioSocketRunnable implements Runnable { 
            @Override
            public void run() { 
                try { 
                    String server = "localhost";
                    int servPort = 8000;
                    String msg = "Hello!";
                    SocketChannel clntChan = SocketChannel.open();
                    clntChan.configureBlocking(false);
                    if (!clntChan.connect(new InetSocketAddress(server, servPort))) { 
                        while (!clntChan.finishConnect()) { 
                            System.out.print(".");
                        }
                    }
                    System.out.print("\n");
                    ByteBuffer writeBuf = ByteBuffer.wrap(msg.getBytes());
                    ByteBuffer readBuf = ByteBuffer.allocate(msg.getBytes().length);
                    int totalBytesRcvd = 0;
                    int bytesRcvd;
                    while (totalBytesRcvd < msg.getBytes().length) { 
                        if (writeBuf.hasRemaining()) { 
                            clntChan.write(writeBuf);
                            Thread.sleep(6000);
                            clntChan.write(writeBuf);
                        }
                        if ((bytesRcvd = clntChan.read(readBuf)) == -1) { 
                            throw new SocketException("Connection closed prematurely");
                        }
                        totalBytesRcvd += bytesRcvd;
                    }
                    System.out.println("Received: " + new String(readBuf.array(), 0, totalBytesRcvd));
                    clntChan.close();
    
                } catch (IOException | InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
    
    }

启动服务端后再启动客户端，最后服务端输出：

socket: Socket[addr=/127.0.0.1,port=59058,localport=8000], spend: 1ms
    socket: Socket[addr=/127.0.0.1,port=59065,localport=8000], spend: 0ms
    socket: Socket[addr=/127.0.0.1,port=59063,localport=8000], spend: 0ms
    socket: Socket[addr=/127.0.0.1,port=59062,localport=8000], spend: 1ms
    socket: Socket[addr=/127.0.0.1,port=59064,localport=8000], spend: 0ms
    socket: Socket[addr=/127.0.0.1,port=59066,localport=8000], spend: 0ms
    socket: Socket[addr=/127.0.0.1,port=59060,localport=8000], spend: 0ms
    socket: Socket[addr=/127.0.0.1,port=59061,localport=8000], spend: 1ms
    socket: Socket[addr=/127.0.0.1,port=59059,localport=8000], spend: 0ms
    socket: Socket[addr=/127.0.0.1,port=59067,localport=8000], spend: 0ms

服务端并不会一直阻塞。

## AIO ##

`AIO`即`Asynchronized IO`，是完全异步的IO。在NIO中，虽然不用一直等待IO，IO操作准备好时再通知线程处理，但是IO操作本身仍然是同步的，只是节省了等待的时间，IO操作本身消耗的时间是仍然可以感知到的。

AIO不同于NIO的地方在于：IO操作完成后再给线程发出一个通知，因此AIO完全不会阻塞，所以相应地，IO操作完成后会执行一个回调函数，由系统触发。

一个AIO实现的Server：

public class AioServer { 
    
        public final static int PORT = 8088;
        private static AsynchronousServerSocketChannel server;
    
        public static void main(String[] args) throws IOException, InterruptedException { 
            asynchronousServer();
    
            start();
    
            blocking();
    
        }
    
        private static void blocking() throws InterruptedException { 
    
            while (true){ 
                System.out.println("Server running...");
                Thread.sleep(1000L);
            }
    
        }
    
        private static void asynchronousServer() throws IOException { 
            server = AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(PORT));
        }
    
        private static void start(){ 
            System.out.println("Server listen on " + PORT);
            server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() { 
    
                ByteBuffer buffer = ByteBuffer.allocate(1024);
    
                @Override
                public void completed(AsynchronousSocketChannel result, Object attachment) { 
                    System.out.println(Thread.currentThread().getName() + " on completed...");
                    Future<Integer> writeResult = null;
                    try { 
    
                        buffer.clear();
                        result.read(buffer).get(100, TimeUnit.SECONDS);
                        buffer.flip();
                        writeResult = result.write(buffer);
    
                    }catch (Exception e){ 
                        e.printStackTrace();
                    }finally { 
                        try { 
                            server.accept(null,this);
                            writeResult.get();
                            result.close();
                        }catch (Exception e){ 
                            e.printStackTrace();
                        }
                    }
                }
    
                @Override
                public void failed(Throwable exc, Object attachment) { 
                    System.out.println(Thread.currentThread().getName() + " failed...");
                }
            });
        }
    
    }

这个Server就简单地把客户端传来的数据返回给客户端，与NIO不同的点在于：`accept()`方法不再阻塞，而是立马返回，并注册一个`CompletionHandler`实例，处理客户端连接，其中的两个方法分别会在成功或失败的情况下被调用；`read()/write()`方法不再阻塞，所以想要等待返回结果必须使用`Future.get()`；整个线程都是异步且是驻守后台的，如果想要服务始终保持，需要主动阻塞。

Client：

public class AioClient { 
    
        public static void main(String[] args) throws IOException, InterruptedException { 
            AsynchronousSocketChannel client = AsynchronousSocketChannel.open();
            client.connect(new InetSocketAddress("localhost", AioServer.PORT), null, new CompletionHandler<Void, Object>() { 
                @Override
                public void completed(Void result, Object attachment) { 
                    client.write(ByteBuffer.wrap("Hello!".getBytes()), null, new CompletionHandler<Integer, Object>() { 
                        @Override
                        public void completed(Integer result, Object attachment) { 
                            try{ 
                                ByteBuffer buffer = ByteBuffer.allocate(1024);
                                client.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() { 
                                    @Override
                                    public void completed(Integer result, ByteBuffer attachment) { 
                                        buffer.flip();
                                        System.out.println("From Server: " + new String(buffer.array()));
                                        try { 
                                            client.close();
                                        } catch (IOException e) { 
                                            e.printStackTrace();
                                        }
                                    }
    
                                    @Override
                                    public void failed(Throwable exc, ByteBuffer attachment) { 
                                        System.out.println("Read failed...");
                                    }
                                });
                            }catch (Exception e){ 
                                e.printStackTrace();
                            }
                        }
    
                        @Override
                        public void failed(Throwable exc, Object attachment) { 
                            System.out.println("Write failed...");
                        }
                    });
    
                }
    
                @Override
                public void failed(Throwable exc, Object attachment) { 
                    System.out.println("Connect failed...");
                }
            });
    
            Thread.sleep(1000L);
        }
    
    }

先后启动Server和Client，IO操作的结果全都定义在`CompletionHandler`里，IO真正执行完毕后会触发这个`Handler`的方法，服务端完成数据处理和对客户端回复，客户端发送数据并接收服务端的应答。

[Disruptor_3_Disruptor]: https://blog.csdn.net/twypx/article/details/80398886
[disruptor]: https://github.com/LMAX-Exchange/disruptor
[CompletableFuture]: https://www.jianshu.com/p/6f3ee90ab7d3