Java高并发（七）——Future模式

超、凢脫俗 2022-04-06 15:16 324阅读 0赞

大家想下，多线程处理提高性能的根本本质在哪？其实就是将串行的处理步骤进行并行的处理，其实总时间是没有缩短的。也就是以前一个人干活需要10个小时，而十个人干同样的活需要1小时，从而缩短处理时间。但是如果干活有先后限制怎么办？例如工作中：测试前必须编码，编码前必须设计，设计前必须需求分析，分析前……如何提高这种情况的性能呢？或者说是如何让这中情况下的线程更加充分利用呢？Future模式——异步调用。好，在总结Future模式前，我们来先看一篇文章[https://www.cnblogs.com/cz123/p/7693064.html][https_www.cnblogs.com_cz123_p_7693064.html]。

**一，流程对比**：对上边厨房做饭的例子，进行用Future模式的前后执行流程对比，从图中可以看出，其实就将能够并行的处理步骤通过异步调用进行了并行，并行不了的只能串行。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70][]

二**，Future模式**：

1，Future的核心结构图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70 1][]

2，看下实现简易代码：

/**
     * 数据接口
     */
    public interface Data {
        /**
         * 获取数据
         * @return
         */
        String getResult();
    }
    
    
    /**
     * RealData类
     */
    public class RealData implements Data {
        protected final String result;
    
        public RealData(String para) {
            StringBuffer sb = new StringBuffer();
            for (int i = 0; i < 10; i++) {
                sb.append(para);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            result = sb.toString();
        }
    
    
        @Override
        public String getResult() {
            return result;
        }
    }
    
    
    /**
     * FutureData类
     */
    public class FutureData implements Data {
        //FutureData是realData的包装
        protected RealData realData = null;
        protected boolean isReady = false;
    
        public synchronized void setRealData(RealData realData) {
            if (isReady) {
                return;
            }
            this.realData = realData;
            isReady = true;
            notifyAll();
        }
    
    
        @Override
        public synchronized String getResult() {
            while (!isReady) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return realData.getResult();
        }
    }
    
    
    
    public class Client {
    
        public Data request(final String queryStr) {
            final FutureData futureData = new FutureData();
            //单起个线程进行数据处理
            new Thread() {
                @Override
                public void run() {
                    RealData realData = new RealData(queryStr);
                    futureData.setRealData(realData);
                }
            }.start();
            //立即返回
            return futureData;
        }
    
    
        public static void main(String[] args) {
            Client client = new Client();
    
            Data data = client.request("ljhname");
            System.out.println("请求完毕");
    
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            System.out.println("数据" + data.getResult());
        }
    }

三**，JDK中的Future模式**：JDK已经帮我们准备了一套完整的实现，我们可以利用其进行非常方便的实现功能。

1，先把上边的例子改为使用jdk的：

public class RealData implements Callable<String> {
        private String para;
    
        public RealData (String para){
            this.para = para;
        }
    
        @Override
        public String call() throws Exception {
            StringBuffer sb = new StringBuffer();
            for (int i = 0; i < 10; i++) {
                sb.append(para);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
    
            return sb.toString();
        }
    }
    
    
    
    public class FutureMain {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
    
            FutureTask<String> futureTask = new FutureTask<String>(new RealData("ljh"));
            ExecutorService executorService = Executors.newFixedThreadPool(1);
    
            executorService.submit(futureTask);
    
            System.out.println("请求完毕");
    
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            System.out.println("数据=" + futureTask.get());
    
        }
    }

2，看下FutureTask的类关系结构图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70 2][]

3，FutureTask源码分析：

/**
         * 一，属性：几个处理状态流程 callable,outcome runner,waiter...
         * The run state of this task, initially NEW.  The run state
         * transitions to a terminal state only in methods set,
         * setException, and cancel.  During completion, state may take on
         * transient values of COMPLETING (while outcome is being set) or
         * INTERRUPTING (only while interrupting the runner to satisfy a
         * cancel(true)). Transitions from these intermediate to final
         * states use cheaper ordered/lazy writes because values are unique
         * and cannot be further modified.
         *
         * Possible state transitions:
         * NEW -> COMPLETING -> NORMAL
         * NEW -> COMPLETING -> EXCEPTIONAL
         * NEW -> CANCELLED
         * NEW -> INTERRUPTING -> INTERRUPTED
         */
        private volatile int state;
        private static final int NEW          = 0;
        private static final int COMPLETING   = 1;
        private static final int NORMAL       = 2;
        private static final int EXCEPTIONAL  = 3;
        private static final int CANCELLED    = 4;
        private static final int INTERRUPTING = 5;
        private static final int INTERRUPTED  = 6;
    
        /** The underlying callable; nulled out after running */
        private Callable<V> callable;
        /** The result to return or exception to throw from get() */
        private Object outcome; // non-volatile, protected by state reads/writes
        /** The thread running the callable; CASed during run() */
        private volatile Thread runner;
        /** Treiber stack of waiting threads */
        private volatile WaitNode waiters;
    
    
        //二，两个构造方法
        /**
         * Creates a {@code FutureTask} that will, upon running, execute the
         * given {@code Callable}.
         *
         * @param  callable the callable task
         * @throws NullPointerException if the callable is null
         */
        public FutureTask(Callable<V> callable) {
            if (callable == null)
                throw new NullPointerException();
            this.callable = callable;
            this.state = NEW;       // ensure visibility of callable
        }
    
        /**
         * Creates a {@code FutureTask} that will, upon running, execute the
         * given {@code Runnable}, and arrange that {@code get} will return the
         * given result on successful completion.
         *
         * @param runnable the runnable task
         * @param result the result to return on successful completion. If
         * you don't need a particular result, consider using
         * constructions of the form:
         * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
         * @throws NullPointerException if the runnable is null
         */
        public FutureTask(Runnable runnable, V result) {
            this.callable = Executors.callable(runnable, result);
            this.state = NEW;       // ensure visibility of callable
        }
    
    
        //三，取消任务
        public boolean cancel(boolean mayInterruptIfRunning) {
            if (!(state == NEW &&
                  UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                      mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
                return false;
            try {    // in case call to interrupt throws exception
                if (mayInterruptIfRunning) {
                    try {
                        Thread t = runner;
                        if (t != null)
                            t.interrupt();
                    } finally { // final state
                        UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                    }
                }
            } finally {
                finishCompletion();
            }
            return true;
        }
    
       //四，是否已经取消
        public boolean isCancelled() {
            return state >= CANCELLED;
        }
    
    
        //五，取的返回对象，get()直到去的为止，get(long timeout, TimeUnit unit)有取的等待时间，超过则报超时异常
        /**
         * @throws CancellationException {@inheritDoc}
         */
        public V get() throws InterruptedException, ExecutionException {
            int s = state;
            if (s <= COMPLETING)
                s = awaitDone(false, 0L);
            return report(s);
        }
    
        /**
         * @throws CancellationException {@inheritDoc}
         */
        public V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
            if (unit == null)
                throw new NullPointerException();
            int s = state;
            if (s <= COMPLETING &&
                (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
                throw new TimeoutException();
            return report(s);
        }
    
        //六，是否完成
        public boolean isDone() {
            return state != NEW;
        }
    
        //七，执行
        public void run() {
            if (state != NEW ||
                !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                             null, Thread.currentThread()))
                return;
            try {
                Callable<V> c = callable;
                if (c != null && state == NEW) {
                    V result;
                    boolean ran;
                    try {
                        result = c.call();
                        ran = true;
                    } catch (Throwable ex) {
                        result = null;
                        ran = false;
                        setException(ex);
                    }
                    if (ran)
                        set(result);
                }
            } finally {
                // runner must be non-null until state is settled to
                // prevent concurrent calls to run()
                runner = null;
                // state must be re-read after nulling runner to prevent
                // leaked interrupts
                int s = state;
                if (s >= INTERRUPTING)
                    handlePossibleCancellationInterrupt(s);
            }
        }

好，Future模式先这样吧，主要理解其通过线程异步处理的过程，当然JDK8中又增加了CompletableFuture，功能更加强大，后边我会总结上。继续中……

[https_www.cnblogs.com_cz123_p_7693064.html]: https://www.cnblogs.com/cz123/p/7693064.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70]: /images/20220406/1f45c1f0bac24fd0a1813bd9342aaf73.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70 1]: /images/20220406/5dc205ee81234b939813819221e631f1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70 2]: /images/20220406/ceeb29704e204a0e9564cced99083e18.png