FutureTask

傳統(tǒng)的通過繼承Thread或者實現(xiàn)runnable接口，不能獲得該線程的執(zhí)行情況。通過callable和future可以獲得任務(wù)的執(zhí)行結(jié)果。

• Future是一個接口，它定義的方法：
  1、boolean cancel(boolean mayInterruptIfRunning);
  2、 boolean isCancelled();
  3、 boolean isDone();
  4、 V get() throws InterruptedException, ExecutionException;
• 簡單說明一下接口定義
  1、 boolean cancel(boolean mayInterruptInRunning) 取消一個任務(wù)，并返回取消結(jié)果。參數(shù)表示是否中斷線程。
  2、 boolean isCancelled() 判斷任務(wù)是否被取消
  3、 Boolean isDone() 　　 判斷當前任務(wù)是否執(zhí)行完畢，包括正常執(zhí)行完畢、執(zhí)行異?；蛘呷蝿?wù)取消。
  4、 V get() 獲取任務(wù)執(zhí)行結(jié)果，任務(wù)結(jié)束之前會阻塞。V get(long timeout, TimeUnit unit) 在指定時間內(nèi)嘗試獲取執(zhí)行結(jié)果。若超時則拋出超時異常

圖片.png

• FutureTask實現(xiàn)了RunnableFuture接口，則RunnableFuture接口繼承了Runnable接口和Future接口，所以FutureTask既能當做一個Runnable直接被Thread執(zhí)行，也能作為Future用來得到Callable的計算結(jié)果。
• Future使用案例

package com.alan.concurrency.example.aqs;


import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

@Slf4j
public class FutureExample {


    static class MyCallable implements Callable<String>{


        //Callable相比runnable可以返回任意類型參數(shù)
        @Override
        public String call() throws Exception {
            log.info("do something in callable");
            Thread.sleep(5000);
            return "done !!!!";
        }
    }


    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        //通過Future接口的實現(xiàn)可以獲得callable中返回的參數(shù)
        Future<String> future = executorService.submit(new MyCallable());
        log.info("do something in main");
        Thread.sleep(1000);
        String result = future.get();
        log.info("result:{}",result);
    }
}

• FutureTask使用案例

package com.alan.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

@Slf4j
public class FutureTaskExample {

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

        FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                log.info("do something in callable");
                Thread.sleep(5000);
                return "done !!!!";
            }
        });

        new Thread(futureTask).start();
        log.info("do something in main");
        Thread.sleep(1000);
        String result = futureTask.get();
        log.info("result:{}",result);

    }
}

Fork/Join框架（JDK1.7）

• ForkJoin是Java7提供的原生多線程并行處理框架，其基本思想是將大人物分割成小任務(wù)，最后將小任務(wù)聚合起來得到結(jié)果。它非常類似于HADOOP提供的MapReduce框架，只是MapReduce的任務(wù)可以針對集群內(nèi)的所有計算節(jié)點，可以充分利用集群的能力完成計算任務(wù)。ForkJoin更加類似于單機版的MapReduce。

  
  
  圖片.png
• 1+2+3+。。。。+100案例

package com.alan.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;

@Slf4j
public class ForkJoinTaskExample extends RecursiveTask<Integer> {

    public static final int threshold = 2;
    private int start;
    private int end;

    public ForkJoinTaskExample(int start, int end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        int sum = 0;

        //如果任務(wù)足夠小就計算任務(wù)
        boolean canCompute = (end - start) <= threshold;
        if (canCompute) {
            for (int i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            // 如果任務(wù)大于閾值，就分裂成兩個子任務(wù)計算
            int middle = (start + end) / 2;
            ForkJoinTaskExample leftTask = new ForkJoinTaskExample(start, middle);
            ForkJoinTaskExample rightTask = new ForkJoinTaskExample(middle + 1, end);

            // 執(zhí)行子任務(wù)
            leftTask.fork();
            rightTask.fork();

            // 等待任務(wù)執(zhí)行結(jié)束合并其結(jié)果
            int leftResult = leftTask.join();
            int rightResult = rightTask.join();

            // 合并子任務(wù)
            sum = leftResult + rightResult;
        }
        return sum;
    }

    public static void main(String[] args) {
        ForkJoinPool forkjoinPool = new ForkJoinPool();

        //生成一個計算任務(wù)，計算1+2+3+4
        ForkJoinTaskExample task = new ForkJoinTaskExample(1, 100);

        //執(zhí)行一個任務(wù)
        Future<Integer> result = forkjoinPool.submit(task);

        try {
            log.info("result:{}", result.get());
        } catch (Exception e) {
            log.error("exception", e);
        }
    }
}

BlockingQueue(線程安全的)

圖片.png

圖片.png

圖片.png

圖片.png