java并发编程的艺术笔记第六章——java并发容器和框架

会跳舞的机器人

发布于 2018-09-21 16:11:00

3180

发布于 2018-09-21 16:11:00

文章被收录于专栏：会跳舞的机器人

1、ConcurrentHashMap的实现原理与使用

1.1、为什么使用ConcurrentHashMap

HashMap非线程安全
HashTable读写都需要加锁，效率低下
ConcurrentHashMap的锁分段技术可以提高并发效率

1.2、ConcurrentHashMap的结构

ConcurrentHashMap由Segment数组结构和HashEntry数组结构组成，Segement是一种可重入锁，在ConcurrentHashMap扮演着锁的角色；HashEntry用于存储键值对数据，一个ConcurrentHashMap中包含一个Segment数组，它是数组和链表结构。一个Segment里包含一个HashEntry数组，当对HashEntry数组进行修改操作时必须要获取它对应的Segment锁。

1.3、ConcurrentHashMap的初始化

1.4、定位Segment

通过散列算法定位Segment，散列冲突

2、ConcurrentLinkedQueue

并发编程中实现线程安全的队列有两种方式，一种是阻塞队列，一种是非阻塞队列，非阻塞的实现方式可以通过CAS方式来实现。

ConcurrentLinkedQueue是一个基于链接节点的无界安全队列。它采用先进先出的方式对节点进行排序，当我们添加一个元素的时候，它会添加至队列的队尾，当我们获取元素的时候，它会返回队列头部的元素。

3、java中的阻塞队列

阻塞队列支持阻塞的添加/移除元素的方法。支持阻塞的插入的意思是：当队列已满时，队列会阻塞插入队列的线程，直到队列有空位；支持阻塞的移除的意思是：当队列为空时，队列会阻塞移除队列元素的线程，直到队列中有新的元素添加进来。

阻塞队列场用于生产/消费者模式，生产者是向队列中添加元素的线程，消费者是从队列中获取元素的线程，而阻塞队列在其中充当着容器的角色。阻塞队列的插入和移除有四种操作方式，详情请参考文档。

java中有7中阻塞队列，分别是：

ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列
LinkedBlockingQueue：一个由链表结构组成的有界阻塞队列。
PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列。
DelayQueue：一个支持延时获取元素的无界阻塞队列。
SynchronousQueue：一个不存储元素的阻塞队列。
LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。

着重介绍下DelayQueue，它可以运用在以下业务场景：

缓存系统的设置：可以用DelayQueue保存元素的有效期，使用一个线程无限循环查询DelayQueue，一旦能从DelayQueue中获取到元素，则说明缓存有效期到了。
使用DelayQueue保存当天将会执行的任务和时间，一旦从DelayQueue获取到任务，就开始执行，TimeQueue就是使用DelayQueue来实现的。

在公司有这么一个业务场景：订单支付后要给商户发送相应的通知，针对同一条通知记录，如果是第一次发，则需要等待的时间是0分钟，第二次发则需要等待1分钟，第三次发则需要等待3分钟，即发送次数每+1，则需要等待的时长也要相应的增加，那使用DelayQueue就能很好的实现这个功能了。以下是参考实现：

package main.java.com.robot.demo;

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/**
 * @author: 会跳舞的机器人
 * @date: 2017/8/17 14:43
 * @description: 通知任务
 */
public class NotifyTask implements Delayed, Runnable {
    /**
     * 任务名称
     */
    private String notifyTaskName;

    /**
     * 执行时间，单位：毫秒
     */
    private long executeTime;

    public NotifyTask(String notifyTaskName, long executeTime) {
        this.notifyTaskName = notifyTaskName;
        this.executeTime = executeTime;
    }


    /**
     * 获取延迟时间
     *
     * @param unit
     * @return 返回当前元素还需要延长多久时间
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(executeTime - System.currentTimeMillis(), unit.MILLISECONDS);
    }

    /**
     * 用来指定元素的顺序,让延时时间最长的放在队列的末尾
     *
     * @param o
     * @return
     */
    @Override
    public int compareTo(Delayed o) {
        NotifyTask notifyTask = (NotifyTask) o;
        return executeTime > notifyTask.executeTime ? 1 : (executeTime < notifyTask.executeTime ? -1 : 0);
    }

    @Override
    public void run() {
        System.out.println("当前时间毫秒数：" + System.currentTimeMillis() + "，线程：" + this.toString() + "正在执行");

    }

    @Override
    public String toString() {
        return "NotifyTask{" +
                "notifyTaskName='" + notifyTaskName + '\'' +
                ", executeTime=" + executeTime +
                '}';
    }
}

测试：

package main.java.com.robot.demo;

import java.util.Random;
import java.util.concurrent.DelayQueue;

/**
 * @author: 会跳舞的机器人
 * @date: 2017/8/17 14:52
 * @description: 延迟队列DelayQueue测试
 */
public class NotifyTaskTest {
    /**
     * 通知任务存放的延迟队列
     */
    private static DelayQueue<NotifyTask> tasks = new DelayQueue<>();


    public static void main(String[] args) {
        Random random = new Random();
        for (int i = 0; i < 5; i++) {
            // 随机产生一个秒数
            int seconds = random.nextInt(5);
            NotifyTask notifyTask = new NotifyTask("任务" + i, System.currentTimeMillis() + (seconds * 1000));
            tasks.put(notifyTask);
        }
        while (true) {
            NotifyTask notifyTask = tasks.poll();
            if (notifyTask != null) {
                notifyTask.run();
            }
            // 如果队列中的元素全部被取完,则跳出循环
            if (tasks.size() == 0) {
                break;
            }
        }
    }
}

控制台输出：

当前时间毫秒数：1502953649855，线程：NotifyTask{notifyTaskName='任务1', executeTime=1502953649855}正在执行
当前时间毫秒数：1502953649855，线程：NotifyTask{notifyTaskName='任务3', executeTime=1502953649855}正在执行
当前时间毫秒数：1502953650855，线程：NotifyTask{notifyTaskName='任务0', executeTime=1502953650855}正在执行
当前时间毫秒数：1502953651855，线程：NotifyTask{notifyTaskName='任务2', executeTime=1502953651855}正在执行
当前时间毫秒数：1502953651855，线程：NotifyTask{notifyTaskName='任务4', executeTime=1502953651855}正在执行

从输出中可以看出任务的执行时间都是我们创建任务的时候指定的时间。

4、Fork/Join框架

4.1、什么是Fork/Join框架

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

我们再通过Fork和Join这两个单词来理解一下Fork/Join框架。Fork就是把一个大任务切分为若干子任务并行的执行，Join就是合并这些子任务的执行结果，最后得到这个大任务的结果。比如计算1+2+…+10000，可以分割成10个子任务，每个子任务分别对1000个数进行求和，最终汇总这10个子任务的结果。

4.2、工作窃取算法

工作窃取（work-stealing）算法是指某个线程从其他队列里窃取任务来执行。那么，为什么需要使用工作窃取算法呢？假如我们需要做一个比较大的任务，可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应。比如A线程负责处理A 队列里的任务。但是，有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。

工作窃取算法的优点：充分利用线程进行并行计算，减少了线程间的竞争。

工作窃取算法的缺点：在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且该算法会消耗了更多的系统资源，比如创建多个线程和多个双端队列。

4.3、使用Fork/Join框架

让我们通过一个简单的需求来使用Fork/Join框架，需求是：计算1+2+3+4的结果。

package main.java.com.robot.demo;

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

/**
 * @author: 会跳舞的机器人
 * @date: 2017/8/17 15:45
 * @description: Fork/Join框架简单demo
 */
public class CountTask extends RecursiveTask<Integer> {
    /**
     * 阈值
     */
    private static final int THRESHOLD = 2;

    private int start;

    private int end;

    public CountTask(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 midel = (end - start) / 2;
            CountTask leftTask = new CountTask(start, midel);
            CountTask rightTask = new CountTask(midel + 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
        CountTask task = new CountTask(1, 4);
        // 执行一个任务
        Future<Integer> result = forkJoinPool.submit(task);
        try {
            System.out.println(result.get());
        } catch (InterruptedException e) {
        } catch (ExecutionException e) {
        }
    }
}