[Java并发系列]Java并发容器和框架

使用Java并发容器/框架可以提高并发程序的开发效率,设计出更加优秀的并发程序。Java并发容器/框架还是位于java.util.concurrent包中。常用的并发容器有如下:

  • ConcurrentHashMap
  • ConcurrentLinkedQueue
  • 阻塞队列

1. ConcurrentHashMap 的实现原理与使用(基于1.7)

ConcurrentHashMap 是线程安全且高效的HashMap,它可以在保证线程安全的同时可以保证高效的操作。

1. HashMap在并发环境中的问题

① 在并发环境中,HashMap的put操作后可能会导致get操作产生死循环 因为HashMap是使用拉链法解决哈希冲突,即通过链表的方式存储散列到数组相同位置的值。当在并发环境中使用HashMap时,可能会导致数组中的链表形成循环链表,在之后的get操作用会使用e = e.next去判断链表中是否还有元素,而形成链表之后,e = e.next操作就会无限循环,Infinite Loop的错误就会出现。

② 并发环境下,操作HashMap的put方法可能会导致元素丢失 当在多线程下put操作时,执行addEntry(hash, key, value, i),如果有产生哈希碰撞,导致两个线程得到同样的bucketIndex去存储,就可能会出现覆盖丢失的情况:

2. 并发环境中如何获取线程安全的哈希表结构
  • 使用Collections类中的synchronizedMap(Map m)静态方法获取线程安全的Map,然后在此map上再进行操作;
 Map m = Collections.synchronizedMap(new HashMap());
  • 使用HashTable,HashTable是通过在方法或者被调用的方法上添加synchronized关键字来保证线程安全。如果在线程竞争比较激烈的情况下,HashTable的效率可能是非常低下的,因为在HashTable中,线程都必须要竞争同一把锁,正因为如此,当一个线程访问HashTable的同步方法时,其他线程只能进入阻塞或者是轮询状态。
  • 使用ConcurrentHashMap,ConcurrentHashMap 使用锁分段技术,既能保证Map的线程安全,也能保证Map的效率。
3. ConcurrentHashMap的结构

ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成。Segment是一种可重入锁,是由数组和链表组成,在ConcurrentHashMap中,只有一个Segment数组,一个Segment中只包含一个HashEntry数组,每个HashEntry是一个链表结构的元素;当对HashEntry数组中的数据进行修改时,必须首先获取到与它对应的Segment锁。

4. ConcurrentHashMap的常用操作源码解读(1.8)

本机jdk环境为1.8 ,就以1.8的源码来分析,之后会有1.7与1.8中的比较总结

一、put操作

public V put(K key, V value) {        
    return putVal(key, value, false);
}
    
final V putVal(K key, V value, boolean onlyIfAbsent) {        
        //键值对不能为空
        if (key == null || value == null) throw new NullPointerException();        //两次hash,以减少冲突,使元素均匀分布
        int hash = spread(key.hashCode());        
        int binCount = 0; //对数组table进行无条件自循迭代
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh; //判断数组是否为空,如果为空则再初始化(此地方与1.7中实现不同,使用懒汉模式)
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();            //判断i位置是否有元素,若没有的话,则使用无锁化CAS的方法进行元素的插入
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {                //使用CAS插入元素
                if (casTabAt(tab, i, null,                             new Node<K,V>(hash, key, value, null)))                    break;        // no lock when adding to empty bin
            }            
            //如果需要扩容,则先进行扩容操作
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);            
            else {
                V oldVal = null;                
            //上述条件都不满足,则说明存在hash冲突,则需要进行加锁操作,锁住链表或红黑树的头结点
                synchronized (f) {                    
                    if (tabAt(tab, i) == f) {                        
                        if (fh >= 0) {//表示该节点位置是链表结构
                            binCount = 1;//循环该节点位置的链表
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;  //先判断是否有相同的key,如果有则直接覆盖原来的value值
                                if (e.hash == hash &&((ek = e.key) == key ||(ek != null && key.equals(ek)))) {
                                    oldVal = e.val;                                    
                                    if (!onlyIfAbsent)
                                         e.val = value;                                    
                                    break;
                                }                                
                            //没有重复的key,则将元素插入到链表的尾部
                            Node<K,V> pred = e;                                
                            if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,value, null);                                    
                                break;
                                }
                            }
                        }                        
                        else if (f instanceof TreeBin) {//表示该节点位置是红黑树结构
                            Node<K,V> p;
                            binCount = 2;//插入到红黑树中
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,value)) != null) {
                                oldVal = p.val;                                
                            if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }                
               
               if (binCount != 0) {                
               //当链表上的元素个数大于8时,就会将链表转换为红黑树
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);                    
               if (oldVal != null)                        
                   return oldVal;                    
               break;
                }
            }
        }
        addCount(1L, binCount);        
       return null;
    }

put操作流程:

  1. 先对key进行两次hash计算;
  2. 检查table是否已经初始化,如果没有初始化,那么就先进行初始化;
  3. 如果没有hash冲突,那就直接使用CAS插入元素;
  4. 如果需要扩容,就先进行扩容操作;
  5. 如果产生了Hash冲突,那就对链表或者红黑树头结点进行加锁,将Key/value插入到链表尾部或者按照红黑树结构进行插入;
  6. 在插入结束之后,需要判断table该位置的链表长度是否大于8,如果大于8时,就会将链表转换为红黑树结构。
  7. 最后统计size;

二、**get操作**

    public V get(Object key) {
        Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;  
        //对key进行两次hash计算,定位到在table中的位置
        int h = spread(key.hashCode());        
        if ((tab = table) != null && (n = tab.length) > 0 &&(e = tabAt(tab, (n - 1) & h)) != null) {//直接读取首节点位置的元素
            if ((eh = e.hash) == h) {//如果首节点元素的hash值与key的两次hash的值相等,那么就通过对象的equals()方法,来比较key与首节点的key值是否相等,如果相等,则返回;
               if ((ek = e.key) == key || (ek != null && key.equals(ek)))                    
                    return e.val;
            }            //hash值为负值表示正在扩容,这个时候查的是ForwardingNode的find方法来定位到nextTable进行查找,
            else if (eh < 0)
                return (p = e.find(h, key)) != null ? p.val : null;//既不是首节点也不是ForwardingNode,那就往下遍历
            while ((e = e.next) != null) {                
                if (e.hash == h &&
                    ((ek = e.key) == key || (ek != null && key.equals(ek))))                    
                return e.val;

            }
        }        
        return null;
    }

get操作流程:

  • 计算key的hash值,定位到table的索引位置,如果是首节点,那么就直接返回;
  • 如果遇到正在扩容,那么就调用标志正在扩容的ForwardingNode的find方法,查找该节点,匹配就直接返回;
  • 如果上述情况都不合符,那么就遍历该链表上的节点,使用key.equals(ek)去匹配key值,匹配则返回;

三、 size操作

public int size() {        
    long n = sumCount();        
    return ((n < 0L) ? 0 : (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :(int)n);
    }   final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;        
    long sum = baseCount;        
    if (as != null) {        
    //将CounterCell与baseCount累加
            for (int i = 0; i < as.length; ++i) {                
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }        
     return sum;
    }
5. 1.7与1.8中ConcurrentHashMap的不同
  • 1.8中使用HashEntry+Synchronized+CAS+红黑树控制并发,1.7中使用Segment+Reentrantlock+HashEntry来实现;1.7中锁的粒度是基于Segment的,而1.8中锁的粒度是基于HashEntry(首节点)的,所以说1.8中锁的粒度更低,
  • 1.8中使用Synchronized来进行同步,所以不需要在分段,也就不需要Segment分段的数据结构,降低实现的复杂度。
  • 1.8中使用红黑树来优化链表,使用长链表进行遍历时一个很耗时的过程,而基于红黑树来查找,则会大大提高查找的速率。

2. ConcurrentLinkedQueue

在并发环境中,如果要得到一个线程安全的队列,可以通过两种方式获取:一种是通过阻塞的方式获取,另一种非阻塞的方式则是通过CAS的方式实现;而ConcurrentLinkedQueue则就是使用非阻塞的方式实现的线程安全的队列。 ConcurrentLinkedQueue是一个基于链表的无界线程安全队列。当添加一个元素时,它会将元素添加到队列的尾部,当获取一个元素时,它会从队列的头部返回一个元素。

3. Java中的阻塞队列

阻塞队列:当队列满时,队列就会阻塞向队列中插入插入元素的线程,直到队列不满;当队列为空时,会阻塞获取元素的线程,直到队列非空; 阻塞队列经常用于生产者消费者的场景,生产者是向队列中插入元素的线程,消费者是从队列中取出元素的线程。所以阻塞队列就是生产者用来存放元素、消费者用来消费元素的容器。

  • ArrayBlockingQueue 使用数组实现的有界阻塞队列
  • LinkedBlockingQueue 使用链表实现的有界阻塞队列,默认构造函数如下:
//
 public LinkedBlockingQueue() {        
        this(Integer.MAX_VALUE);
    }
  • PriorityBlockingQueue 支持优先级排序的无界阻塞队列
  • DelayQueue 使用优先级队列实现的无界阻塞队列
  • SynchronousQueue 不存储元素的阻塞队列,每个put操作必须要等待一个take操作,否则是不能添加元素的;
  • LinkedTransferQueue 使用链表实现的无界阻塞队列
  • LinkedBlockingDeque 使用链表组成的双向阻塞队列

阻塞队列的实现原理:就是使用通知模式,当生产者往满的队列中添加元素时,就会被阻塞,当消费者从队列中消费元素之后,就会通知生产者当前的队列可用;当消费者想要在一个为空的队列中消费时,就会被阻塞,当生产者在此队列中生产元素之后,就会通知消费者此队列可用。

原文发布于微信公众号 - 瞎说开发那些事(jsj201501)

原文发表时间:2017-10-23

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏java一日一条

【Java并发编程】使用wait/notify/notifyAll实现线程间通信的几点重要说明

在 Java中,可以通过配合调用Object对象的wait()方法和notify()方法或notifyAll()方法来实现线程间的通信。在线程中调用 wait(...

603
来自专栏你不就像风一样

Java多线程编程核心技术(二)对象及变量的并发访问

本文主要介绍Java多线程中的同步,也就是如何在Java语言中写出线程安全的程序,如何在Java语言中解决非线程安全的相关问题。阅读本文应该着重掌握如下技术点:

703
来自专栏机器学习从入门到成神

Java使用增强for循环和迭代器遍历Map集合

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/sinat_35512245/articl...

761
来自专栏xingoo, 一个梦想做发明家的程序员

回调函数

  有点类似模板的功能,可以使用函数指针作为参数,当调用函数时,使用void *进行传递参数,细致比较时,再用int *之类的进行强制转换。回调函数,其实就是在...

2318
来自专栏Jack-Cui

234. Palindrome Linked List(Linked List-Easy)

Given a singly linked list, determine if it is a palindrome. Follow up: Could yo...

18710
来自专栏java工会

Java多线程实现的三种方式

1505
来自专栏Java Edge

Java8集合源码解析-Hashtable源码剖析1 概述2 源码解析rehash方法3 总结

2656
来自专栏个人分享

多线程模拟实现生产者/消费者模型 (借鉴)

 在生产者/消费者模型中,生产者Producer负责生产数据,而消费者Consumer负责使用数据。多个生产者线程会在同一时间运行,生产数据,并放到内存中一个共...

953
来自专栏编程坑太多

python进程和线程概述

1145
来自专栏小灰灰

JDK容器学习之LinkedHashMap(二):迭代遍历的实现方式

LinkedHashMap 如何保障有序的遍历 前一篇《JDK容器学习之LinkedHashMap (一):底层存储结构分析》 中介绍了LinkedHashM...

2037

扫码关注云+社区