FutureTask原理

使用FutureTask可以返回当前线程的结果。

可以让线程阻塞或者是唤醒

LockSupport与Wait/notify

FutureTask简单用法

public class Test001 implements Callable<String> {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        FutureTask<String> stringFutureTask = new FutureTask<>(new Test001());

        new Thread(stringFutureTask).start();

        String result = stringFutureTask.get();

        System.out.println(result);

    }

  

    @Override

    public String call() throws Exception {

        try {

            Thread.sleep(3000);

            System.out.println(Thread.currentThread().getName() + ",发送短信完成。");

        } catch (Exception e) {

  

        }

        return "发送短信完成";

    }

  

  

}

手写FurureTask 方式一

public interface MyCallable<V> {

    V call();

}

public class MayiktTask<V> implements Runnable {

    private MyCallable<V> myCallable;

    private V result;

    private Thread cuThread;

  

    public MayiktTask(MyCallable myCallable) {

        this.myCallable = myCallable;

    }

  

    @Override

    public void run() {

        result =  myCallable.call();

        LockSupport.unpark(cuThread);

    }

  

    public V get() {

        if (result != null) {

            return result;

        }

        cuThread = Thread.currentThread();

        LockSupport.park(cuThread);

        return result;

    }

}

MayiktTask<String> stringMayiktTask = new MayiktTask<>(new Test002());

  new Thread(stringMayiktTask).start();

String s = stringMayiktTask.get();

System.out.println(s);

手写FurureTask 方式二

public class MayiktTask<V> implements Runnable {

    private MyCallable<V> myCallable;

    private V result;

    private Thread cuThread;

  

    private Object lock = new Object();

  

    public MayiktTask(MyCallable myCallable) {

        this.myCallable = myCallable;

    }

  

    @Override

    public void run() {

        result = myCallable.call();

        synchronized (lock) {

            lock.notify();

        }

  //        LockSupport.unpark(cuThread);

    }

  

    public V get() {

        if (result != null) {

            return result;

        }

        cuThread = Thread.currentThread();

  //        LockSupport.park(cuThread);

        synchronized (lock) {

            try {

                lock.wait();

            } catch (Exception e) {

  

            }

        }

        return result;

    }

}

Fork Join 原理解读

并发编程发展

Java 1 支持thread，synchronized。

Java 5 引入了 thread pools， blocking queues, concurrent collections，locks, condition queues。

Java 7 加入了fork-join库。

Java 8 加入了 parallel streams。并行流：

基本设计思想

Fork/Join是Java7提供的并行执行任务的框架，是一个把大任务分割成若干小任务，最终汇总小任务的结果得到大任务结果的框架

小任务可以继续不断拆分n多个小任务。

并发编程第12篇，FutureTask原理

基本伪代码实现

if(任务很小）{

直接计算得到结果

}else{

分拆成N个子任务

调用子任务的fork()进行计算

调用子任务的join()合并计算结果

}

工作窃取机制

将一个比较大的任务，拆分成n多个不同的子任务，一直到不可以拆分为止。

例如：将一个大的任务，拆分成n多个子任务，每个任务中对应一个独立的队列。

由于每个线程处理的速度不一样，如果先执行完任务的队列的线程，窃取其他没有

执行完任务队列。

Fork JOIN用法

使用Fork join计算总和

public class ForkJoinDemo extends RecursiveTask<Long> {

    // 最小分隔单位

    private long max = 200;

    private long start;

    private long end;

  

    public ForkJoinDemo(Long start, Long end) {

        this.start = start;

        this.end = end;

    }

  

  

    @Override

    protected Long compute() {

        Long sum = 0l;

        if (end - start < max) {

            System.out.println(Thread.currentThread().getName() + ",start:" + start + ",end:" + end);

            for (Long i = start; i <= end; i++) {

                sum += i;

            }

        } else {

            // 400+1 200 1-200,201-400

            long l = (end + start) / 2;

            ForkJoinDemo left = new ForkJoinDemo(start, l);

            left.fork();

            ForkJoinDemo rigt = new ForkJoinDemo(l + 1, end);

            rigt.fork();

            left.join();

            rigt.join();

            try {

                sum = left.get() + rigt.get();

            } catch (InterruptedException e) {

                e.printStackTrace();

            } catch (ExecutionException e) {

                e.printStackTrace();

            }

        }

        return sum;

    }

  

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        ForkJoinPool forkJoinPool = new ForkJoinPool();

        ForkJoinDemo forkJoinDemo = new ForkJoinDemo(1l, 400l);

        ForkJoinTask<Long> submit = forkJoinPool.submit(forkJoinDemo);

        System.out.println(submit.get());

  

    }

}

使用Fork JOIN异步群发短信

public class ForkJoinSms extends RecursiveAction {

    /**

     * 存放手机号码

     */

    private List<String> phones;

  

    private int start;

    private int end;

    // 1000

    private int max = 100;

  
    public ForkJoinSms(int start, int end) {

        this.start = start;

        this.end = end;

    }

  

    @Override

    protected void compute() {

        if (end - start < max) {

            System.out.println(Thread.currentThread().getName() + ",start:" + start + ",end:" + end);

        } else {

            int l = (end + start) / 2;

            ForkJoinSms left = new ForkJoinSms(start, l);

  

            ForkJoinSms rigt = new ForkJoinSms(l + 1, end);

            left.fork();

            rigt.fork();

        }

  

    }

 

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        ForkJoinSms forkJoinSms = new ForkJoinSms(1, 1000);

        ForkJoinPool forkJoinPool = new ForkJoinPool();

        forkJoinPool.invoke(forkJoinSms);


    }

}

每一行代码都有它的涵义，多问一句为什么；别怕，理清思路，一切代码都是数据的流动和转化，耐心一点，慢慢积累！一起加油！！！