J.U.C 组件扩展

FutureTask

传统的通过继承Thread或者实现runnable接口，不能获得该线程的执行情况。通过callable和future可以获得任务的执行结果。

• Future是一个接口，它定义的方法：
  1、boolean cancel(boolean mayInterruptIfRunning);
  2、 boolean isCancelled();
  3、 boolean isDone();
  4、 V get() throws InterruptedException, ExecutionException;
• 简单说明一下接口定义
  1、 boolean cancel(boolean mayInterruptInRunning) 取消一个任务，并返回取消结果。参数表示是否中断线程。
  2、 boolean isCancelled() 判断任务是否被取消
  3、 Boolean isDone() 　　 判断当前任务是否执行完毕，包括正常执行完毕、执行异常或者任务取消。
  4、 V get() 获取任务执行结果，任务结束之前会阻塞。V get(long timeout, TimeUnit unit) 在指定时间内尝试获取执行结果。若超时则抛出超时异常

图片.png

• FutureTask实现了RunnableFuture接口，则RunnableFuture接口继承了Runnable接口和Future接口，所以FutureTask既能当做一个Runnable直接被Thread执行，也能作为Future用来得到Callable的计算结果。
• 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可以返回任意类型参数
        @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接口的实现可以获得callable中返回的参数
        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提供的原生多线程并行处理框架，其基本思想是将大人物分割成小任务，最后将小任务聚合起来得到结果。它非常类似于HADOOP提供的MapReduce框架，只是MapReduce的任务可以针对集群内的所有计算节点，可以充分利用集群的能力完成计算任务。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;

        //如果任务足够小就计算任务
        boolean canCompute = (end - start) <= threshold;
        if (canCompute) {
            for (int i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            // 如果任务大于阈值，就分裂成两个子任务计算
            int middle = (start + end) / 2;
            ForkJoinTaskExample leftTask = new ForkJoinTaskExample(start, middle);
            ForkJoinTaskExample rightTask = new ForkJoinTaskExample(middle + 1, end);

            // 执行子任务
            leftTask.fork();
            rightTask.fork();

            // 等待任务执行结束合并其结果
            int leftResult = leftTask.join();
            int rightResult = rightTask.join();

            // 合并子任务
            sum = leftResult + rightResult;
        }
        return sum;
    }

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

        //生成一个计算任务，计算1+2+3+4
        ForkJoinTaskExample task = new ForkJoinTaskExample(1, 100);

        //执行一个任务
        Future<Integer> result = forkjoinPool.submit(task);

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

BlockingQueue(线程安全的)

图片.png
图片.png
图片.png
图片.png