ConcurrentHashMap的简单使用

Java PV:

ConcurrentHashMap的介绍

ConcurrentHashMap是一个线程安全，并且是一个高效的HashMap,ConcurrentMap，它是一个接口，是一个能够支持并发访问的java.util.map集合,在原有java.util.map接口基础上又新提供了4种方法，进一步扩展了原有Map的功能

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public interface ConcurrentMap<K, V> extends Map<K, V> {
    //插入元素
    V putIfAbsent(K key, V value);
    //移除元素
    boolean remove(Object key, Object value);
    //替换元素
    boolean replace(K key, V oldValue, V newValue);
    //替换元素
    V replace(K key, V value);
}

(1) putIfAbsent: 与原有put方法不同的是,putIfAbsent方法中如果插入的key相同,则不替换原有的value值

(2) remove: 与原有remove方法不同的是,新remove方法中增加了对value的判断,如果要删除的key–value不能与Map中原有的key–value对应上,则不会删除该元素

(3) replace(K,V,V): 增加了对value值的判断,如果key–oldValue能与Map中原有的key–value对应上,才进行替换操作

(4) replace(K,V): 与上面的replace不同的是,此replace不会对Map中原有的key–value进行比较,如果key存在则直接替换

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
27
28
    private static Logger log = Logger.getLogger(ConcurrentMapDemo.class);
    @Test
    public void test() {
        ConcurrentHashMap<Object, Object> concurrentHashMap = new ConcurrentHashMap<Object, Object>();
        concurrentHashMap.put("name", "tom");
        log.info("concurrentHashMap>>>>>   " + concurrentHashMap.toString());
//        putIfAbsent: 与原有put方法不同的是,putIfAbsent方法中如果插入的key相同,则不替换原有的value值
        concurrentHashMap.putIfAbsent("name", "lucy");
        log.info("concurrentHashMap>>>>>   " + concurrentHashMap.toString());
//        remove: 与原有remove方法不同的是,新remove方法中增加了对value的判断,如果要删除的key--value不能与Map中原有的key--value对应上,
//        则不会删除该元素
        concurrentHashMap.remove("name", "lucy");
        log.info("concurrentHashMap>>>>>   " + concurrentHashMap.toString());
//        replace(K,V,V): 增加了对value值的判断,如果key--oldValue能与Map中原有的key--value对应上,才进行替换操作
        concurrentHashMap.replace("name", "lucy", "newLucy");
        log.info("concurrentHashMap>>>>>   " + concurrentHashMap.toString());
//        replace(K,V): 与上面的replace不同的是,此replace不会对Map中原有的key--value进行比较,如果key存在则直接替换
        concurrentHashMap.replace("name", "lucy");
        log.info("concurrentHashMap>>>>>   " + concurrentHashMap.toString());
    }

其实,对于ConcurrentMap来说,我们更关注Map本身的操作,在并发情况下是如何实现数据安全的。在java.util.concurrent包中,ConcurrentMap的实现类主要以ConcurrentHashMap为主,接下来,我们具体来看下

从线程安全的角度来说,HashTable已经是一个线程安全的HashMap,说起ConcurrentHashMap,就不得不先提及下HashMap在线程不安全的表现

HashMap遍历删除

HashMap或者ArrayList边遍历边删除数据会报java.util.ConcurrentModificationException异常

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@Test
public void test() {
    Map<Long, String> mReqPacket = new HashMap<Long, String>();
    for (long i = 0; i < 15; i++) {
        mReqPacket.put(i, i + "");
    }
    for (Entry<Long, String> entry : mReqPacket.entrySet()) {
        long key = entry.getKey();
        String value = entry.getValue();
        if (key < 10) {
            mReqPacket.remove(key);
        }
    }
    for (Entry<Long, String> entry : mReqPacket.entrySet()) {
        log.info(entry.getKey() + " " + entry.getValue());
    }
}

运行结果: 

1
2
3
4
5
6
7
8
9
10
11
java.util.ConcurrentModificationException
	at java.util.HashMap$HashIterator.nextNode(HashMap.java:1437)
	at java.util.HashMap$EntryIterator.next(HashMap.java:1471)
	at java.util.HashMap$EntryIterator.next(HashMap.java:1469)
	at map.ConcurrentMapDemo.test1(ConcurrentMapDemo.java:119)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:389)
Process finished with exit code -1

使用迭代器删除元素,没有问题,不会出现异常 : 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@Test
public void test() {
  Map<Long, String> mReqPacket = new HashMap<Long, String>();
    for (long i = 0; i < 15; i++) {
        mReqPacket.put(i, i + "");
    }
    for (Iterator<Entry<Long, String>> iterator = mReqPacket.entrySet().iterator(); iterator.hasNext();) {
        Entry<Long, String> entry = iterator.next();
        long key = entry.getKey();
        if (key < 10) {
            iterator.remove();
        }
    }
    for (Entry<Long, String> entry : mReqPacket.entrySet()) {
        log.info(entry.getKey() + " " + entry.getValue());
    }
}

ConcurrentHashMap允许一边更新、一边遍历，也就是说在Iterator对象遍历的时候，ConcurrentHashMap也可以进行remove,put操作，且遍历的数据会随着remove,put操作产出变化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Test
public void test1() {
    Map<Long, String> conMap = new ConcurrentHashMap<Long, String>();
    for (long i = 0; i < 15; i++) {
        conMap.put(i, i + "");
    }
    for (Entry<Long, String> entry : conMap.entrySet()) {
        long key = entry.getKey();
        if (key < 10) {
            conMap.remove(key);
        }
    }
    for (Entry<Long, String> entry : conMap.entrySet()) {
        log.info(entry.getKey() + " " + entry.getValue());
    }
}

ConcurrentHashMap多线程操作

对ConcurrentHashMap边遍历边删除或者增加操作不会产生异常(可以不用迭代方式删除元素)，因为其内部已经做了维护，遍历的时候都能获得最新的值。即便是多个线程一起删除、添加元素也没问题

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
static Map<Long, String> conMap = new ConcurrentHashMap<Long, String>();
public static void main(String[] args) throws InterruptedException {
    for (long i = 0; i < 5; i++) {
        conMap.put(i, i + "");
    }
    Thread thread = new Thread(new Runnable() {
        public void run() {
            for (Iterator<Entry<Long, String>> iterator = conMap.entrySet().iterator(); iterator.hasNext();) {
                Map.Entry<Long, String> entry = iterator.next();
                log.info(entry.getKey() + " - " + entry.getValue());
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    });
    Thread thread2 = new Thread(new Runnable() {
        public void run() {
            conMap.put(100l, "100");
            log.info("ADD:" + 100);
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    });
    thread.start();
    thread2.start();
    Thread.sleep(3000);
    log.info("--------");
    for (Map.Entry<Long, String> entry : conMap.entrySet()) {
        log.info(entry.getKey() + " " + entry.getValue());
    }
}

输出结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
2017-11-27 22:03:45,123 [Thread-0] INFO  [map.ConcurrentMapDemo] - 0 - 0
2017-11-27 22:03:45,123 [Thread-1] INFO  [map.ConcurrentMapDemo] - ADD:100
2017-11-27 22:03:45,248 [Thread-0] INFO  [map.ConcurrentMapDemo] - 1 - 1
2017-11-27 22:03:45,357 [Thread-0] INFO  [map.ConcurrentMapDemo] - 2 - 2
2017-11-27 22:03:45,466 [Thread-0] INFO  [map.ConcurrentMapDemo] - 3 - 3
2017-11-27 22:03:45,575 [Thread-0] INFO  [map.ConcurrentMapDemo] - 4 - 4
2017-11-27 22:03:45,685 [Thread-0] INFO  [map.ConcurrentMapDemo] - 100 - 100
2017-11-27 22:03:48,135 [main] INFO  [map.ConcurrentMapDemo] - --------
2017-11-27 22:03:48,135 [main] INFO  [map.ConcurrentMapDemo] - 0 0
2017-11-27 22:03:48,135 [main] INFO  [map.ConcurrentMapDemo] - 1 1
2017-11-27 22:03:48,135 [main] INFO  [map.ConcurrentMapDemo] - 2 2
2017-11-27 22:03:48,135 [main] INFO  [map.ConcurrentMapDemo] - 3 3
2017-11-27 22:03:48,135 [main] INFO  [map.ConcurrentMapDemo] - 4 4
2017-11-27 22:03:48,135 [main] INFO  [map.ConcurrentMapDemo] - 100 100
Process finished with exit code 0

ConcurrentHashMap结构

我们来了解下ConcurrentHashMap的整体结构，这样有利于我们快速理解源码。我们就在此进行回顾下

HashMap的整体结构

HashMap底层使用数组和链表，实现哈希表结构。插入的元素通过散列的形式分布到数组的各个角标下；当有重复的散列值时，便将新增的元素插入在链表头部，使其形成链表结构，依次向后排列。

下面是，ConcurrentHashMap的结构:

与HashMap不同的是，ConcurrentHashMap中多了一层数组结构，由Segment和HashEntry两个数组组成。其中Segment起到了加锁同步的作用，而HashEntry则起到了存储K.V键值对的作用

在ConcurrentHashMap中，每一个ConcurrentHashMap都包含了一个Segment数组，在Segment数组中每一个Segment对象则又包含了一个HashEntry数组，而在HashEntry数组中，每一个HashEntry对象保存K-V数据的同时又形成了链表结构，此时与HashMap结构相同

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
27
28
29
30
31
32
33
34
private void writeObject(java.io.ObjectOutputStream s)
        throws java.io.IOException {
        // For serialization compatibility
        // Emulate segment calculation from previous version of this class
        int sshift = 0;
        int ssize = 1;
        while (ssize < DEFAULT_CONCURRENCY_LEVEL) {
            ++sshift;
            ssize <<= 1;
        }
        int segmentShift = 32 - sshift;
        int segmentMask = ssize - 1;
        @SuppressWarnings("unchecked")
        Segment<K,V>[] segments = (Segment<K,V>[])
            new Segment<?,?>[DEFAULT_CONCURRENCY_LEVEL];
        for (int i = 0; i < segments.length; ++i)
            segments[i] = new Segment<K,V>(LOAD_FACTOR);
        s.putFields().put("segments", segments);
        s.putFields().put("segmentShift", segmentShift);
        s.putFields().put("segmentMask", segmentMask);
        s.writeFields();
        Node<K,V>[] t;
        if ((t = table) != null) {
            Traverser<K,V> it = new Traverser<K,V>(t, t.length, 0, t.length);
            for (Node<K,V> p; (p = it.advance()) != null; ) {
                s.writeObject(p.key);
                s.writeObject(p.val);
            }
        }
        s.writeObject(null);
        s.writeObject(null);
        segments = null; // throw away
    }

在多线程中，每一个Segment对象守护了一个HashEntry数组，当对ConcurrentHashMap中的元素修改时，在获取到对应的Segment数组角标后，都会对此Segment对象加锁，之后再去操作后面的HashEntry元素，这样每一个Segment对象下，都形成了一个小小的HashMap，在保证数据安全性的同时，又提高了同步的效率。只要不是操作同一个Segment对象的话，就不会出现线程等待的问题

分享

Powered by Hexo and Hexo-theme-hiker

Copyright © 2018 - 2019 ZhouXu'Blog All Rights Reserved.

UV : | PV :