Java并发编程的艺术笔记-Java并发容器和框架

1.ConcurrentHashMap的实现原理与使用

1.1 为什么要使用ConcurrentHashMap

• 线程不安全的HashMap：HashMap在并发执行put操作时会引起死循环(JDK1.8之前的问题)
  • 因为多线程会导致HashMap的Entry链表形成环形数据结构，一旦形成环形数据结构，Entry的next节点永远不为空，就会产生死循环获取Entry(原因是因为JDK1.7链表加入节点是头插法)

final HashMap<String, String> map = new HashMap<>(2);
Thread t = new Thread(new Runnable() {
    @Override
    public void run() {
        for (int i = 0; i < 10000; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    map.put(UUID.randomUUID().toString(), "");
                }
            }, "ftf" + i).start();
        }
    }
}, "ftf");
t.start();
t.join();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

• 效率低下的HashTable：HashTable容器使用synchronized来保证线程安全

  • 线程1使用put进行元素添加，线程2不但不能使用put方法，也不能使用get方法(访问HashTable的线程都必须竞争同一把锁)

• ConcurrentHashMap使用锁分段技术：

  • 假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，则多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争

1.2 ConcurrentHashMap的结构

• 由Segment数组结构和HashEntry数组结构组成

• Segment是一种可重入锁(ReentrantLock)

• HashEntry用于存储键值对数据

• Segment的结构和HashMap类似，是一种数组和链表结构

• 一个Segment里包含一个HashEntry数组，每个HashEntry是一个链表结构的元素

• 当对HashEntry数组的数据进行修改时，须先获得与它对应的Segment锁

1.3 定位Segment

• 在插入和获取元素的时候，ConcurrentHashMap须先通过散列算法定位到Segment
• ConcurrentHashMap会使用算法对元素的hashCode进行一次再散列(减少Hash冲突，使元素均匀分布在不同的Segment上)

1.4 ConcurrentHashMap的操作

• get操作：
  • 整个get过程不需要加锁，除非读到的值是空才会加锁重读
  • 将要使用的共享变量都定义成volatile类型(get方法中自需要读不需要写这些共享变量)

public V get(Object key) {
    // 1.先经过一次再散列(一次再散列)
    int hash = hash(key.hashCode());
    // 2.使用这个散列值通过散列运算定位到Segment，再通过散列算法定位到元素(两次散列定位)
    return segmentFor(hash).get(key, hash);
}

1
2
3
4
5
6
7

• put操作：

  • 因为需要对共享变量进行写入操作，所以在操作共享变量时必须加锁

  • 插入操作需要经历两个步骤：

    • 是否需要扩容：在插入元素前会先判断Segment里的HashEntry数组是否超过容量，如果超过阈值，则对数组进行扩容(HashMap是先插入再判断扩容，很可能扩容后就没有元素要插入了)

    • 如何扩容：创建一个容量是原来容量两倍的数组，然后将原数组里的元素进行再散列后插入到新的数组里(只对某个segment进行扩容)

• size操作：先尝试2次通过不锁住Segment的方式来统计各个Segment大小，如
  果统计的过程中，容器的count发生了变化，则再采用加锁的方式来统计所有Segment的大小

---

2.Java中的阻塞队列

2.1 什么是阻塞队列

• 支持两个附加操作的队列：

  • 支持阻塞的插入方法：当队列满时，队列会阻塞插入元素的线程，直到队列不满
  • 支持阻塞的移除方法：在队列为空时，获取元素的线程会等待队列变为非空

• 常用于生产者和消费者的场景，即生产者用来存放元素、消费者用来获取元素的容器

2.2 Java里的阻塞队列

• ArrayBlockingQueue：

  • 用数组实现的有界阻塞队列
  • 默认情况下不保证线程公平的访问队列，可以指定为公平队列(底层使用ReentrantLock(fair))

• LinkedBlockingQueue：

  • 用链表实现的有界阻塞队列

• PriorityBlockingQueue：

  • 支持优先级的无界阻塞队列
  • 不能保证同优先级元素的顺序

• DelayQueue：

  • 支持延时获取元素的无界阻塞队列

  • 在创建元素时可以指定多久才能从队列中获取当前元素(只有在延迟期满时才能从队列中提取元素)

• SynchronousQueue：

  • 不存储元素的阻塞队列(意味着每一个put操作必须等待一个take操作，否则不能继续添加元素)
  • 支持公平访问队列(默认采用非公平性策略访问队列)

• LinkedTransferQueue：

  • 由链表结构组成的无界阻塞TransferQueue队列
  • 相较于其他阻塞队列多了tryTransfer和transfer方法
    • transfer：把生产者传入的元素立刻transfer（传输）给消费者
    • tryTransfer：试探生产者传入的元素是否能直接传给消费者

• LinkedBlockingDeque：

  • 由链表结构组成的双向阻塞队列

  • 在多线程同时入队时减少了一半的竞争

---

3.Fork/Join框架

3.1 什么是Fork/Join框架

• Fork：把一个大任务切分为若干子任务并行的执行
• Join：合并这些子任务的执行结果，最后得到大任务的结果

3.2 工作窃取算法

• 指某个线程从其他队列里窃取任务来执行(因为子任务会被分到不同的队列，并且由不同线程负责，而不同线程执行速度不同)

• 为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列(被窃取任务线程从双端队列的头部拿任务，窃取任务的线程从双端队列的尾部拿任务)

• 优点：

  • 充分利用线程进行并行计算，减少了线程间的竞争

• 缺点：

  • 在某些情况下还是存在竞争(比如双端队列里只有一个任务时)
  • 会消耗了更多的系统资源(比如创建多个线程和多个双端队列)

3.3 Fork/Join框架的设计

• 分割任务：fork类来把大任务分割成子任务

• 执行任务并合并结果：

  • 分割的子任务分别放在双端队列里，然后几个启动线程分别从双端队列里获取任务执行
  • 子任务执行完的结果都统一放在一个队列里，启动一个线程从队列里拿数据，然后合并这些数据

3.4 使用Fork/Join框架

@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;
    CountTask leftTask = new CountTask(start, middle);
    CountTask rightTask = new CountTask(middle + 1, end);
    // 执行子任务
    leftTask.fork();
    rightTask.fork();
    // 等待子任务执行完，并得到其结果
    int leftResult=leftTask.join();
    int rightResult=rightTask.join();
    // 合并子任务
    sum = leftResult + rightResult;
    }
    return sum;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

---