ConcurrentHashMap JDK1.8源码深入剖析

ConcurrentHashMap JDK1.8源码深入剖析(一)

阅读前建议掌握的知识

  1. CAS操作
  2. hashMap的原理
  3. 并发相关锁的知识
  4. 并发中原子性，可见性，重排序的概念

特性

  • 底层实现原理为数组+链表+红黑树
  • 当链表大小阀值超过8时转换为红黑树
  • 大量使用CAS操作来实现无锁
  • 并发扩容

主要属性

TREEIFY_THRESHOLD,UNTREEIFY_THRESHOLD
1
2
static final int TREEIFY_THRESHOLD = 8;
static final int UNTREEIFY_THRESHOLD = 6;

大于8时,链表转树阀值。

小于6时树转链表，仅在扩容tranfer时才可能树转链表。

sizeCtl
1
private transient volatile int sizeCtl;

hash表初始化或扩容时的一个控制位标识量。负数代表正在进行初始化或扩容操作。

  • -1代表正在初始化。
  • -N 表示有N-1个线程正在进行扩容操作。
  • 正数或0代表hash表还没有被初始化，这个数值表示初始化或下一次进行扩容的大小。
MOVED
1
2
3
4
/*
  * Encodings for Node hash fields. See above for explanation.
  */
 static final int MOVED     = -1; // hash for forwarding nodes

Node的hash值，为 -1时代表该Node为ForwardingNode

TREEBIN
1
static final int TREEBIN   = -2;// hash for roots of trees

树根节点的hash值

MAX_RESIZERS
1
2
3
4
5
/**
 * The maximum number of threads that can help resize.
 * Must fit in 32 - RESIZE_STAMP_BITS bits.
 */
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;

并发扩容时最大允许的help线程总数

Node

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
static class Node<K,V> implements Map.Entry<K,V> {
        final int hash;
        final K key;
        //让value为volatile类型，让其有可见性
        volatile V val;
        //
        volatile Node<K,V> next;

        Node(int hash, K key, V val, Node<K,V> next) {
...
        }

...

        //禁止直接setValue
        public final V setValue(V value) {
            throw new UnsupportedOperationException();
        }

  ...

        Node<K,V> find(int h, Object k) {
            Node<K,V> e = this;
            if (k != null) {
                do {
                    K ek;
                    if (e.hash == h &&
                        ((ek = e.key) == k || (ek != null && k.equals(ek))))
                        return e;
                } while ((e = e.next) != null);
            }
            return null;
        }
    }

节点是ConcurrentHashMap的核心内部类，他的设计原理是不允许call setValue来直接改变它的值,因为put需要用CAS来实现无锁线程安全操作。同时提供了find方法来通过hash值和key寻找链表中的节点，该方法被map.get()调用，被子类overriden，来实现不同类型节点的查询方法。

关键的三个CAS方法

1
2
3
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
    return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}

tabAt方法通过UNsafe的CAS操作来获得在i下标位置的Node

1
2
3
4
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                                    Node<K,V> c, Node<K,V> v) {
    return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}

对i下标位置上的Node进行CAS的set操作，如果有其他线程正在访问则会返回false，ASHIFT是指tab[i]中第i个元素在相对于数组第一个元素的偏移量，而ABASE就算第一数组的内存素的偏移地址,((long)i << ASHIFT) + ABASE就算i最后的地址。

1
2
3
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
    U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}

利用volatile方法set节点位置的值,注意setTabAt总是在被锁定的代码块中才调用。

Put方法

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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
final V putVal(K key, V value, boolean onlyIfAbsent) {
        //不允许key或value为null
        if (key == null || value == null) throw new NullPointerException();
        int hash = spread(key.hashCode());
        int binCount = 0;
        //经典CAS操作，不set成功死不罢休
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
            //初始化table
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
            //如果节点为null，直接用cas方法尝试set,若成功直接break跳出CAS循环。
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                if (casTabAt(tab, i, null,
                             new Node<K,V>(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            }
            //如果是forwardingNode节点，代表有其他线程正在扩容，于是helpTransfer辅助扩容
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            //这里意味者节点下是链表或是树，通过synchronized锁住f表头，则锁住了整个链表或树
            else {
                V oldVal = null;
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                    //判断是链表还是树，fh>=0则为链表
                        if (fh >= 0) {
                            binCount = 1;
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                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;
                }
            }
        }
        //更新map大小，+1
        addCount(1L, binCount);
        return null;
    }

putVal的设计核心思想是

  • 不允许key或value为null。
  • 用spread计算key的hash值再用类似取模的方法来确定在table数组中的位置。
  • 若有需要则初始化table。
  • 若该节点为null，无哈希碰撞，则使用CAS来进行set操作。
  • 若该节点被标记为forward，则代表有其他线程正在进行扩容，当前线程也会辅助进行扩容操作。
  • 若该节点为链表或树，代表哈希碰撞，则使用synchronized来锁住桶，然后进行set操作。注意若在该节点为链表时，会依次向后找到应该存放的位置，若大于阀值8则转化为红黑树，若本来就是树结构则按照树的方法存放。

参考

Redis实现分布式锁

Redis实现分布式锁

分布式锁的一些问题

  1. 并发问题，若多个客户端同时上锁，结果只允许一个客户端成功，其他失败，可以利用redis的SETNX 命令来实现，该命令允许若给定的 key 已经存在，则 SETNX 不做任何动作,设置成功，返回 1 ,设置失败，返回 0 。
  2. 上锁后解锁的问题，可以考虑使用redis key的ttl过期，通过PEXPIRE来设置key的自动过期。若不使用自动过期特性，则需要在锁定结束后使用DEL命令来进行解锁，这种情况有可能发生某个客户端已经crash，于是这个锁会永远锁定，导致其他客户端无法获得锁。

Redis2.6以后新特性

redis2.6.12以后 SET 命令的行为可以通过一系列参数来修改

  1. EX second ：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 。
  2. PX millisecond ：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond 效果等同于 PSETEX key millisecond value 。
  3. NX ：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 。
  4. XX ：只在键已经存在时，才对键进行设置操作。

使用新特性来解决分布式锁

  1. 使用命令 SET key value NX EX second 设置锁的同时为锁设置TTL过期时间，若当锁已经存在时返回nil。

Redisson解决方法

Redisson是redis的一个java客户端,实现了分布式锁和同步器，Redisson的分布式锁RLock Java对象实现了java.util.concurrent.locks.Lock接口，同时还支持自动过期解锁。

1
2
3
4
5
6
7
8
9
10
11
12
13
RLock lock = redisson.getLock("anyLock");
// 最常见的使用方法
lock.lock();

// 支持过期解锁功能
// 10秒钟以后自动解锁
// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS);

// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
...
lock.unlock();

以上是redisson使用锁的方法，它的原理采用了lua脚本的方式，这样能兼容redis 2.6.12之前的版本,锁的核心源码如下RedissonLock.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
        //先检查key是否存在
                  "if (redis.call('exists', KEYS[1]) == 0) then " +
        //key不存在，创建Hash
                      "redis.call('hset', KEYS[1], ARGV[2], 1); " +
        //设置过期时间
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
        //若锁已经存在，并且是同样获得锁的线程调用
                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
        //复用锁，为锁+1
                      "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
        //延长锁定时间
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
         //若无法获得锁，返回锁的剩余时间
                  "return redis.call('pttl', KEYS[1]);",
                    Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }

could not read Username for 'https://github.com': Invalid argument

could not read Username for ‘https://github.com‘: Invalid argument

好久没玩hexo blog了，今天心血来潮想恢复使用，结果发现使用hexo d命令时会报如下错误

1
could not read Username for 'https://github.com': Invalid argument

网上搜索了下，这个问题有以下几个原因：

  1. 如果安装git客户端的时候没有勾选git命令在bash和cmd命令都有效，若是在cmd命令下则因为没有将git添加到windows的path，所以会出现这个错误，一般可以尝试在blog的目录打开git bash再尝试hexo d。

  2. 未将username和email加入到git，可以使用如下命令试试

1
2
git config --global user.name "yourname"
git config --global user.email "youremail"
  1. 检查_config.yml的配置是否正确
1
2
3
4
deploy:
 type: git
 repo: https://github.com/yourname/yourname.github.io.git
 branch: master
  1. 客户端太老，我遇到就是这个问题，即便切换了git bash也仍然报错，于是我升级了新客户端，结果能正常弹出github的用户名和密码输入框，确认后可以正常部署。

Hexo安装教程(一)

什么是Hexo?

Hexo 是一个快速、简洁且高效的博客框架。Hexo 使用 Markdown（或其他渲染引擎）解析文章，在几秒内，即可利用靓丽的主题生成静态网页。

Hexo安装环境

  • Windows7

Hexo安装步骤概览

  1. 安装node.js
  2. 安装github desktop
  3. 安装hexo
hexo

阅读全文 >>

随笔

澳门
没有人是天才，个人的才能是靠自己挖掘的，登上山顶，望望身后已看不见的芸芸众生，回过头来，还需要继续向上。生活本不是闲情逸致，它本就是丑陋的，这个世界不会因为谁的离开就停止送给你恶意。我们唯一要学会的就是战斗，即使遇到困难，我们也要独自无谓的战斗下去。

感悟

如何将UITableViewDelegate和UITableViewDataSource从UITableViewController中分离出来?

大家都知道如果给UITableViewController装载一些数据和控制cell的行为(高度,样式等)都需要指定UITableView的delegate给自身,一般会使用IB或者在viewDidLoad中写上self.tableView.delegate = self;然后实现UITableViewDelegate和UITableViewDataSource中的方法,例如:-tableView:cellForRowAtIndexPath: 等等.
那么如何将实现UITableViewDelegate和UITableViewDataSource的方法脱离出来单独放在一个新的class中呢?

objectiveC

阅读全文 >>

|