[Java知识库]java38-并发框架图-BlockingQueue-ForkJoin-Future-happensbefore-

并发整体框架图

实现一个计数器（体现原子性）

方式1：使用count++

public class Counter{
	private int count;
	public void increase(){
		count++;
	}

问题
该代码可能在多线程条件下出现问题，因为count++是非原子的。
count++ 实际上等于三个操作：读数据，加1，写回数据。
解决
为了防止多线程下访问increase方法会报错，所以给increase方法加锁。
count变量修改了其他线程可能看不到，所以就加个volatile关键字吧。

方式2：加synchronized 和 volatile

public class LockCounter{
	private volatile count;
	public synchronized void increase(){
		count++;
	}
}

问题
加锁会影响效率，可以考虑使用原子操作类的形式。

方式3：AtomicInteger

public class AtomicCounter{
	private AtomicInteger count = new AtomicInteger(0);
	public void increase(){
		count.incrementAndGet();
	}
}

线程的并发和同步

Fork 和 Join

JUC中提供了Fork/Join的并行计算框架，用来处理分治的情况。
分治的思想 = 分而治之，把复杂的问题分解成相似的子问题，然后子问题再分子问题，知道问题分到很简单不必再划分为止，然后层层返回问题的结果。
分治分为两个阶段：分解任务 和 合并结果

第一个阶段：分解任务ForkJoinPool

public class ForkJoinPool extends AbstractExecutorService
把任务分解为一个个小任务直至小任务可以简单的计算返回结果；（Fork分解任务）
ForkJoinPool治理分治任务的线程池，它有多个任务队列（区别于ThreadPoolExecutor线程池），通过ForkJoinPool的invoke、submit、execute提交任务的时候会根据一定规则分配给不同的任务队列（该队列是双端队列）。
ForkJoinPool 有一个机制，当某个工作线程对应消费的任务队列空闲的时候它会去别的忙的任务队列的尾部分担任务过来执行，这种执行又被称之为窃取线程。因为这个性质，所以采用双端队列。
窃取线程如何保证不和被窃取任务的线程冲突。队列的绑定工作线程都从队列头部取任务进行执行，窃取线程会从别的队列的尾部取任务执行。
提交任务方法

构造方法
ForkJoinPool(int parallelism) —— 使用指定的并行级别创建一个ForkJoinPool，使用所有其他的参数的默认值。
完整的参数介绍如下：

双端队列
概念：
限定插入和删除操作在表的两端进行的线性表，两端分别称为端点1和端点2，具有队列和栈性质的数据结构。
普通队列：队列头部删除元素、队列尾部添加元素；
双端队列：队列头部添加元素、队列尾部删除元素。

第二个阶段：合并结果

把每个小任务的结果合并返回得到最终结果。（Join合并结果）
ForkJoinTask，分治任务，等同于Runnable。
抽象类，核心方法是fork和join。fork用来异步执行一个子任务，join会阻塞当前线程等待子任务返回。
ForkJoinTask有两个子类：RecursiveAction 和 RecursiveTask 都是抽象类，通过递归来执行分治任务。
每个子类都有compute抽象方法（任务执行的主要计算量），区别在于RecursiveAction没有返回值、RecursiveTast有返回值。

代码思路

1 创建ForkJoinPool，用于执行分治任务的线程池，parallelism = 并行级别，并发线程数
	ForkJoinPool forkjoinpool = new ForkJoinPool(int parallelism);
2 创建具体操作执行的类，需要继承ForkJoinTask类 或者其子类
	public class ForkJoinTaskxhj类 extends RecursiveTask{}
3 创建ForkJoinTaskxhj类的对象
	ForkJoinTaskxhj类 obj = new ForkJoinTaskxhj类（参数）;
4 通过ForkJoinPool对象的invoke方法提交任务并执行
	forkjoinpool.invoke(obj);

案例应用：斐波那契数列

斐波那契数列：1-1-2-3-5-8-13-21-34…
公式：F(1) = 1,F(2) = 1,F(n) = F(n-2) + F(n-1) （n>=3，n为正整数）

package offer2.Test52;

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

public class Test2 {
    public static void main(String[] args) {
        // 治理分治任务的线程池
        ForkJoinPool forkJoinPool = new ForkJoinPool(4); // 最大并发数4
        // 要计算的斐波那契数列 的 第几个元素
        Fibonacci fibonacci = new Fibonacci(20);
        // 任务执行开始 的时间
        long startTime = System.currentTimeMillis();
        // 指定给定的任务，完成返回其结果
        // forkJoinPool对象.invoke方法的参数是ForkJoinTask类，则就是它的对象或者其后代对象
        // 本例中 fibonacci 是ForkJoinTask类的后代对象。
        Integer result = forkJoinPool.invoke(fibonacci);
        // 任务执行完毕 的时间
        long endTime = System.currentTimeMillis();
        System.out.println("Fork/join sum: " + result + " in " + (endTime - startTime) + " ms.");
    }
// 定义Fibonacci数列方法。继承自RecursiveTask泛型类
// Fibonacci类的定义采用的是成员内部类的形式
    static  class Fibonacci extends RecursiveTask<Integer> {
        // 成员变量
        final int n;
        // 构造方法
        Fibonacci(int n) {
            this.n = n;
        }
        // compute抽象方法重写 任务的主要计算量
        @Override
        protected Integer compute() {
            if (n <= 1) {
                return n;
            }
            Fibonacci f1 = new Fibonacci(n - 1);
            f1.fork();
            Fibonacci f2 = new Fibonacci(n - 2);
            return f2.compute() + f1.join();
        }
    }
}

阻塞队列BlockingQueue

概念

• 阻塞队列与普通队列的区别：当队列是空的时候，从队列中获取元素的操作将会阻塞；当队列是满的，往队列里面添加元素的操作会被阻塞。
• 是java.util.concurrent包提供的接口，常用于多线程编程中容纳任务队列。
• 适用阻塞队列的好处：不需要关心什么阻塞线程、什么时候唤醒线程，阻塞队列都帮助我们考虑了。

常用实现类

常用方法

LinkedBlockingQueue

• 使用LinkedBlockingQueue< E > 实现线程同步。
  LinkedBlockingQueue< E > 是一个基于已连接节点的，范围任意的阻塞队列。
  该队列按照先进先出排序元素。
  队列的头部是在队列中时间最长的元素，队列的尾部是在队列中时间最短的元素，新元素插入到队列的尾部。
  获取队列元素的操作只会获取头部元素，如果队列满了或者为空会进入阻塞状态。
• 常用方法：

Future

概述

在JUC包中。
public interface Future< V >
Future模式很好的解决了那些需要返回值的异步调用。
Future表示异步计算的结果。提供方法来检查计算是否完成，等待其完成，并检索计算结果。只有当计算完成之后，才能使用get方法检索结果；如果需要则阻塞，直到准备就绪。
Future模式本质上是代理模式的一种实际应用。
经常使用Future的场景：计算密集场景、处理大数据量、远程方法调用；
当执行一个长时间运行的任务时，使用Future就可以让我们暂时去执行其他任务，等长任务执行完毕后再返回其结果。

Future模式组成部分

图示：

简单案例

package offer2.Test53.tast1;

// 创建Data
public interface Dataxhj {
    // 抽象类
    public String getResult();
}

// 创建继承自Dataxhj接口的真实实现类
public class RealDataXhj implements Dataxhj{
    // protected受保护的变量；
    protected final String result;
    // 带参构造方法
    public RealDataXhj(String para){
        StringBuffer sb = new StringBuffer(para);
        // 假设这里很慢，构造一个RealDataXhj不是一个容易的事
        result = sb.toString();
    }
    @Override
    public String getResult() {
        return result;
    }
}

// future模式的核心
public class FutureData implements Dataxhj {
    // 创建RealDataXhj对象
    protected RealDataXhj realDataXhj = null;
    protected boolean isReady = false;

    // 设置RealDataXhj对象
    public synchronized void setRealDataXhj(RealDataXhj realDataXhj){
        if(isReady){
            return;
        }
        this.realDataXhj = realDataXhj;
        isReady = true;
        // RealDataXhj已经被注入，通知getResult，唤醒线程
        notifyAll();
    }

    // 等待RealDataXhj构造完成
    @Override
    public synchronized String getResult() {
        while(!isReady){
            try {
                // 等待 直到RealDataXhj被注入
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return realDataXhj.result;
        // RealDataXhj类中的result成员变量
    }
}

// 从Client得到DataXhj
public class Client {
   public Dataxhj request(final String queryStr){
       // 创建FutureData对象
       final FutureData future = new FutureData();
       // 创建线程并启动
       new Thread(){
           @Override
           public void run() {
               // RealData构建很慢，在单独的线程中进行
               RealDataXhj realDataXhj = new RealDataXhj(queryStr);
               // setRealData的时候会notify等待这个future上的对阿宁
               future.setRealDataXhj(realDataXhj);
           }
       }.start();
       System.out.println(future.getResult());
       // FutureData会立即返回，不会等待RealDataXhj被构造完
       return future;
   }
}

public class Mainxhj {
    public static void main(String[] args) {
        Client client = new Client();
        //这里会立即返回，因为得到的是FutureData而不是RealData
        Dataxhj dataxhj = client.request("xhjname");
        System.out.println("请求完毕");
        try {
            //这里可以用一个sleep代替了对其他业务逻辑的处理
            //在处理这些业务逻辑的过程中，RealData被创建，从而充分利用了等待时间
            Thread.sleep(2000);
        } catch (InterruptedException e) {
//            e.printStackTrace();
        }
        //使用真实的数据，如果到这里数据还没有准备好，getResult()会等待数据准备完，再返回
        System.out.println("数据 = " + dataxhj.getResult());
    }
}

java中的Future模式

JDK内部有一个Future接口，类似之前案例中的订单。
常用方法：

案例：2022/5/3不是很理解

package offer2.Test53.tast1;

// 创建Data
public interface Dataxhj {
    // 抽象类
    public String getResult();
}

// 创建继承自Dataxhj接口的真实实现类
public class RealDataXhj implements Dataxhj, Callable {
    // protected受保护的变量；
    protected final String result;
    // 带参构造方法
    public RealDataXhj(String para){
        StringBuffer sb = new StringBuffer(para);
        // 假设这里很慢，构造一个RealDataXhj不是一个容易的事
        result = sb.toString();
    }
    @Override
    public String getResult() {
        return result;
    }

    @Override
    public Object call() throws Exception {
        return result;
    }
}

public class Test2 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 创建一个固定长度的线程池
        ExecutorService executor = Executors.newFixedThreadPool(1);
        final Future<?> future = executor.submit((Callable) new RealDataXhj("xhj"));
        System.out.println("请求完毕，数据准备中");
        try {
            // 仍旧可以做额外的数据操作，这是使用sleep代替
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("数据 = " + future.get());
    }
}

happens-before

happens-before是JMM最核心的概念。

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖