多线程设计模式：第五篇 - Future模式和两阶段终止模式

Posted 2021-01-17

一，Future模式

????????Future 的意思是未来，假设有一个方法需要花费很长的时间才能获取运行结果，那么与其一直等待，不如先拿到一份最终数据的模板，即 Future 角色，等过一阵子再通过 Future 角色去获取数据，如果数据已经好了则直接返回，否则就一直等待到有数据返回。

????????这种模式可以用在不是马上需要一个操作的返回值时，这样可以提高程序的响应性，使得一个方法的多个步骤可以并行执行。

????????下面的代码示例用来演示实现 Future 模式：

/**
 * @author koma <[email protected]>
 * @date 2018-11-04
 */
public class Main {
    public static void main(String[] args) {
        Host host = new Host();
        Data data1 = host.request(10, ‘A‘);
        Data data2 = host.request(20, ‘B‘);
        Data data3 = host.request(30, ‘C‘);
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("data1 = "+data1.getContent());
        System.out.println("data2 = "+data2.getContent());
        System.out.println("data3 = "+data3.getContent());
    }
}

public interface Data {
    public abstract String getContent();
}

public class Host {
    //这里使用 ThreadFactory 工厂类来创建线程
    private static final ThreadFactory threadFactory = Executors.defaultThreadFactory();

    public Data request(final int count, final char c) {
        System.out.println("Request BEGIN");

        FutureData futureData = new FutureData();
        threadFactory.newThread(new MakeRealDataTask(count, c, futureData)).start();

        System.out.println("Request END");
        return futureData;
    }
}

public class MakeRealDataTask implements Runnable {
    private final int count;
    private final char c;
    private final FutureData futureData;

    public MakeRealDataTask(int count, char c, FutureData futureData) {
        this.count = count;
        this.c = c;
        this.futureData = futureData;
    }

    @Override
    public void run() {
        RealData realData = new RealData(count, c);
        futureData.setRealData(realData);
    }
}

public class FutureData implements Data {
    private RealData realData = null;
    private boolean ready = false;

    /**
     * 这里的 synchronized 主要是为了实现 wait/notify 语义
     * 本身该类并不需要同步
     *
     */
    public synchronized void setRealData(RealData realData) { //产生实际数据
        if (ready) {
            return;
        }
        this.realData = realData;
        this.ready = true;
        notifyAll();
    }

    @Override
    public synchronized String getContent() { //获取实际数据，当数据还未产生时，则需要持续等待
        while (!ready) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return realData.getContent();
    }
}

public class RealData implements Data {
    private final String content;

    public RealData(int count, char c) {
        System.out.println("RealData BEGIN");
        char[] buffer = new char[count];
        for (int i = 0; i < count; i++) {
            buffer[i] = c;
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("RealData END");
        this.content = new String(buffer);
    }

    @Override
    public String getContent() {
        return content;
    }
}

????????Future 模式和 Thread-Per-Message 模式的区别在于 Future 模式可以从异步执行的线程中获取返回值。

????????Future 模式也可以有变种使用方式，如：当前示例中我们对 Future 返回值的赋值只有一次，而实际上可以赋值多次，这样通过一个返回对象在不同阶段可以获取到不同的返回值。

????????Java JUC 包中提供了用于支持 Future 模式的接口和类。Callable 接口声明了 call() 方法，该方法和 Runnable 的 run() 方法类似，不同之处在于，call() 方法有返回值。Future 接口充当了 Future 角色，声明了 get() 方法用于获取值，设置值的方法则需要对应的实现类去实现，同时还声明了 cancel() 方法，用于中断程序。FutureTask 类实现了 Future 和 Runnable 接口。

????????下面的代码示例，我们利用 juc 包中已有的类来改造我们的示例程序：

public class FutureData extends FutureTask<RealData> implements Data {
    public FutureData(Callable<RealData> callable) {
        super(callable);
    }

    @Override
    public String getContent() {
        String content = null;
        try {
            content = get().getContent();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return content;
    }
}

public class Host {
    private static final ThreadFactory threadFactory = Executors.defaultThreadFactory();

    public Data request(final int count, final char c) {
        System.out.println("Request BEGIN");

        FutureData futureData = new FutureData(new Callable<RealData>() {
            @Override
            public RealData call() throws Exception {
                return new RealData(count, c);
            }
        });
        threadFactory.newThread(futureData).start();

        System.out.println("Request END");
        return futureData;
    }
}

????????在创建 FutureTask 类实例时，Callable 对象作为构造函数参数传递进去，之后当调用 FutureTask 的 run() 方法时，那么构造参数中接收的 Callable 对象的 call() 方法会被执行，call() 方法会同步的获取 call() 方法的返回值，然后通过 FutureTask 的 set() 方法来设置该返回值，如果 call() 方法发生了异常，则调用 FutureTask 的 setException() 方法设置的异常处理函数。之后当我们需要时调用 FutureTask 的 get() 方法就能够获取到 call() 方法的返回值。

????????Executors 框架中通过 submit() 方法实现了 Future 模式，通过框架我们可以更加简单方便的使用 Future 模式，改造后的代码如下：

public class Main {
    public static void main(String[] args) {
        Host host = new Host();
        Future<RealData> data1 = host.request(10, ‘A‘);
        Future<RealData> data2 = host.request(20, ‘B‘);
        Future<RealData> data3 = host.request(30, ‘C‘);
        try {
            Thread.sleep(2000);
            System.out.println("data1 = "+data1.get().getContent());
            System.out.println("data2 = "+data2.get().getContent());
            System.out.println("data3 = "+data3.get().getContent());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace(); //当执行 call() 方法报错时
        }
    }
}

public class Host {
    public Future<RealData> request(final int count, final char c) {
        System.out.println("Request BEGIN");

        //使用 Executors 框架的 Future 模式
        Future<RealData> future = Executors.newFixedThreadPool(1).submit(new Callable<RealData>() {
            @Override
            public RealData call() throws Exception {
                return new RealData(count, c);
            }
        });

        System.out.println("Request END");
        return future;
    }
}

//这里只需要 RealData 类就够了，其它类不再需要

二，两阶段终止模式

????????该模式通常用于优雅的终止线程，它的意思是先执行完终止处理程序再终止线程。

????????我们称线程在进行正常处理时的状态为操作中。在要求停止该线程时，我们发出终止请求，这样线程不会突然终止，而是转为终止处理中状态，然后执行清理工作，完成之后就会真正的终止线程。

????????从操作中变为终止处理中这是终止的第一阶段，在该阶段下，线程不会再进行正常操作，而是只执行清理程序，清理完成之后，就会真正的终止线程，终止处理中状态结束是线程终止的第二阶段。

????????该模式的要点如下：

• 安全的终止线程
• 必定会终止线程
• 发起终止请求后会尽快进行终止处理

????????下面是一个两阶段终止的演示程序：

/**
 * @author koma <[email protected]>
 * @date 2018-11-04
 */
public class Main {
    public static void main(String[] args) {
        try {
            CounterThread counterThread = new CounterThread();
            counterThread.start();

            Thread.sleep(10000);

            counterThread.shutdownRequest();

            counterThread.join(); //主线程等待counterThread线程终止
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class CounterThread extends Thread {
    private long counter = 0;
    private volatile boolean shutdownRequested = false;

    /**
     * 该方法是线程安全的
     *
     * 因为 shutdownRequested 只会被设置成 true 没有别的方法会再把它设置成 false
     * 不存在数据竞争，因此也就允许多线程调用
     *
     */
    public void shutdownRequest() {
        shutdownRequested = true;
        //给线程自己发出中断信号，确保线程在 sleep 或 wait 中也能正常响应终止请求
        interrupt();
    }

    public boolean isShutdownRequested() {
        return shutdownRequested;
    }

    @Override
    public void run() {
        try {
            while (!isShutdownRequested()) {//判断终止请求
                doWork();
            }
        } finally {
            doShutdown();
        }
    }

    private void doWork() {
        counter++;
        System.out.println("doWorker: counter = "+counter);
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
        }
    }

    private void doShutdown() {
        System.out.println("doShutdown: counter = "+counter);
    }
}

????????Executor 框架中的 ExecutorService 类提供了 shutdown() 方法来优雅的终止线程。该类还提供了 isShutdown() 和 isTerminated() 来判断线程是否终止和是否实际停止。其区别在于当线程处于终止处理中状态时，isShutdown() 返回 true，而 isTerminated() 返回 false。

１，中断状态和 InterruptedException 异常

????????当 interrupt() 方法被调用之后，线程就可以被中断了。中断线程这个行为会带来以下结果之一：

????????(1) 线程变为中断状态，反映为＂状态＂
????????(2) 抛出InterruptedException异常，反映为＂控制＂

????????通常情况下会是结果(1)，但是当线程正则 sleep，wait，join 时则会是结果(2)（这时线程不变为中断状态）。

２，中断状态 转换为 InterruptedException 异常

if (Thread.interrupted()) {
    throw new InterruptedException();
}

????????这段代码可以把中断状态转换为异常，其中 if 条件检查的是当前线程的中断状态，同时会清除当前线程的中断状态，若不想清除线程的中断状态，则可以调用 Thread.currentThread().isInterrupted() 方法。

３，InterruptedException异常 转换为 中断状态

try {
    Thread.sleep(1000);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt(); //将捕获到的中断异常转换为中断状态
}

????????这段代码可以把中断异常转换为中断状态，如果不想当前线程的已经被中断这个状态信息丢失的话可以使用这种方式，即线程再次中断一次自己。

４，juc 包与线程同步

????????当我们想让某个线程等待指定的线程终止时，可以调用欲等待线程的 join() 方法，但是由于 join() 方法可以等待的只是线程终止这个一次性操作，当我们想要实现＂等待指定次数的某种操作发生＂这类需求时，则需要借助 juc 包中的 CountDownLatch 类。

????????CountDownLatch 类只能进行倒数，当想多次重复进行线程同步时，则使用 CyclicBarrier 会更加的方便。CyclicBarrier 可以周期性的创建屏障，在屏障解除之前，碰到屏障的线程无法继续前进，屏障解除的条件是到达屏障处的线程数，达到了构造函数指定的个数。也就是说，当指定个数的线程达到屏障处后，屏障就会被解除。

????????下面的代码实现了＂让三个线程处理一项分为０～４共５个阶段的工作＂的功能，我们要求，除非三个线程都完成了第N个阶段，否则哪个线程都不允许进入到第N+1个阶段：

/**
 * @author koma <[email protected]>
 * @date 2018-11-04 
 */
public class Main {
    private static final int THREADS = 3; //工作线程数

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

        //屏障解除时的操作
        Runnable barrierAction = new Runnable() {
            @Override
            public void run() {
                System.out.println("barrier action");
            }
        };

        //用来设定在屏障处等待的线程数量
        CyclicBarrier cyclicBarrier = new CyclicBarrier(THREADS, barrierAction);

        //用来设定等待结束线程的数量
        CountDownLatch countDownLatch = new CountDownLatch(THREADS);

        try {
            for (int i = 0; i < THREADS; i++) {
                executorService.execute(new MyTask(cyclicBarrier, countDownLatch, i));
            }
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            executorService.shutdown();
        }
    }
}

public class MyTask implements Runnable {
    private static final int PAHSE = 5;
    private final CyclicBarrier pahseBarrier;
    private final CountDownLatch downLatch;
    private final int context;
    private static final Random random = new Random(315246);

    public MyTask(CyclicBarrier pahseBarrier, CountDownLatch downLatch, int context) {
        this.pahseBarrier = pahseBarrier;
        this.downLatch = downLatch;
        this.context = context;
    }

    @Override
    public void run() {
        try {
            //任务分 PAHSE 个阶段执行，每个线程在进入到某一阶段之后就等待其它进入该阶段的线程
            //当每个阶段的线程数达到设定的数量时，该阶段屏障解除，所有线程进入到下一阶段
            //pahseBarrier.await() 可以用来循环的创建屏障
            for (int phase = 0; phase < PAHSE; phase++) {
                doPhase(phase);
                pahseBarrier.await();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        } finally {
            downLatch.countDown();
        }
    }

    protected void doPhase(int phase) {
        String name = Thread.currentThread().getName();
        System.out.println(name+"-MyTask BEGIN, context = "+context+", phase = "+phase);
        try {
            Thread.sleep(random.nextInt(3000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println(name+"-MyTask END, context = "+context+", phase = "+phase);
        }
    }
}