千寻

道路很长, 开始了就别停下!

0%

发表于 分类于 阅读次数： Valine：

SQL生命周期：

  1. 应用服务器与数据库服务器建立一个连接
  2. 数据库进程拿到请求sql
  3. 解析并生成执行计划，执行
  4. 读取数据到内存并进行逻辑处理
  5. 通过步骤一的连接，发送结果到客户端
  6. 关掉连接，释放资源
阅读全文 »

发表于 分类于 阅读次数： Valine：

B+树，B树

区别有以下两点：

  1. B+树中只有叶子节点会带有指向记录的指针，而B树则所有节点都带有，在内部节点出现的索引项不会再出现在叶子节点中。
  2. B+树中所有叶子节点都是通过指针连接在一起，而B树不会。

B+树的优点：

  1. 非叶子节点不会带上指向记录的指针，这样，一个块中可以容纳更多的索引项，一是可以降低树的高度。二是一个内部节点可以定位更多的叶子节点。
  2. 叶子节点之间通过指针来连接，范围扫描将十分简单，而对于B树来说，则需要在叶子节点和内部节点不停的往返移动。具体的来讲，如何想扫描一次所有数据，对于b+树来说，可以从因为他们的叶子结点是连在一起的，所以可以横向的遍历过去。而对于b-树来说，就这能中序遍历了。

B树的优点：

对于在内部节点的数据，可直接得到，不必根据叶子节点来定位。

B树长什么样子？

红黑树和B树应用场景有何不同？

为什么要设计红黑树？

先说一下红黑树，红黑树有一个比较复杂的规则，平衡树和红黑树的区别是什么？为什么有了平衡树还要设计出来红黑树？

红黑树的规则：

  • 1）每个结点要么是红的，要么是黑的。
  • 2）根结点是黑的。
  • 3）每个叶结点（叶结点即指树尾端NIL指针或NULL结点）是黑的。
  • 4）如果一个结点是红的，那么它的俩个儿子都是黑的。
  • 5）对于任一结点而言，其到叶结点树尾端NIL指针的每一条路径都包含相同数目的黑结点。

现在想想，我的理解是平衡树（AVL）更平衡，结构上更加直观，时间效能针对读取而言更高，但是维护起来比较麻烦！！！（插入和删除之后，都需要rebalance）。但是，红黑树通过它规则的设定，确保了插入和删除的最坏的时间复杂度是O(log N) 。

设计红黑树的目的，就是解决平衡树的维护起来比较麻烦的问题，红黑树，读取略逊于AVL，维护强于AVL，每次插入和删除的平均旋转次数应该是远小于平衡树。

小结一下：

能用平衡树的地方，就可以用红黑树。用红黑树之后，读取略逊于AVL，维护强于AVL。

红黑树 和 b+树的用途有什么区别？

  1. 红黑树多用在内部排序，即全放在内存中的，STL的map和set的内部实现就是红黑树。

  2. B+树多用于外存上时，B+也被成为一个磁盘友好的数据结构。

为什么b+磁盘友好？

  1. 磁盘读写代价更低
    树的非叶子结点里面没有数据，这样索引比较小，可以放在一个blcok（或者尽可能少的blcok）里面。避免了树形结构不断的向下查找，然后磁盘不停的寻道，读数据。这样的设计，可以降低io的次数。

  2. 查询效率更加稳定
    非终结点并不是最终指向文件内容的结点，而只是叶子结点中关键字的索引。所以任何关键字的查找必须走一条从根结点到叶子结点的路。所有关键字查询的路径长度相同，导致每一个数据的查询效率相当。

  3. 遍历所有的数据更方便
    B+树只要遍历叶子节点就可以实现整棵树的遍历，而其他的树形结构 要中序遍历才可以访问所有的数据。

为什么mysql索引使用B+树而不使用红黑树?

B+树就是为文件存储而生的。如果数据库文件存储在主存中我认为两种结构的查询速度差距不是很大，因为主存的查找速度非常快。

而数据库文件实际存储在磁盘中，定位一行信息需要查找该行文件所在柱面号，磁盘号，扇区号，页号这个阶段是很耗费时间的。

每一次的定位请求意味着要做一次IO操作，也意味着成倍的时间消耗。因此减少IO查询的次数是提高查询性能的关键。而IO的查询次数就是索引树的高度，高度越低查询的次数越少。

同样的结点次数红黑树的高度最多为2log(n+1)，而B+树的高度最多为(logt (n+1)/2)+1,随着t增大高度会更小，IO次数也会减少。

发表于 分类于 阅读次数： Valine：

现在我们来对每一步的细节进行源码分析，在第一步中，符合条件会进行初始化操作，我们来看看initTable（）方法

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
/**
 * Initializes table, using the size recorded in sizeCtl.
 */
private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {//空的table才能进入初始化操作
        if ((sc = sizeCtl) < 0) //sizeCtl<0表示其他线程已经在初始化了或者扩容了，挂起当前线程 
            Thread.yield(); // lost initialization race; just spin
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//CAS操作SIZECTL为-1，表示初始化状态
            try {
                if ((tab = table) == null || tab.length == 0) {

                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;

                    @SuppressWarnings("unchecked")

                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];//初始化

                    table = tab = nt;

                    sc = n - (n >>> 2);//记录下次扩容的大小
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

在第二步中没有hash冲突就直接调用Unsafe的方法CAS插入该元素，进入第三步如果容器正在扩容，则会调用helpTransfer（）方法帮助扩容，现在我们跟进helpTransfer（）方法看看

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
/**
 *帮助从旧的table的元素复制到新的table中
 */
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; int sc;
    if (tab != null && (f instanceof ForwardingNode) &&
        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) { //新的table nextTba已经存在前提下才能帮助扩容
        int rs = resizeStamp(tab.length);
        while (nextTab == nextTable && table == tab &&
               (sc = sizeCtl) < 0) {
            if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                sc == rs + MAX_RESIZERS || transferIndex <= 0)
                break;
            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                transfer(tab, nextTab);//调用扩容方法
                break;
            }
        }
        return nextTab;
    }
    return table;
}

其实helpTransfer（）方法的目的就是调用多个工作线程一起帮助进行扩容，这样的效率就会更高，而不是只有检查到要扩容的那个线程进行扩容操作，其他线程就要等待扩容操作完成才能工作。

既然这里涉及到扩容的操作，我们也一起来看看扩容方法transfer（）

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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    // 每核处理的量小于16，则强制赋值16
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range
    if (nextTab == null) {            // initiating
        try {
            @SuppressWarnings("unchecked")
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];        //构建一个nextTable对象，其容量为原来容量的两倍
            nextTab = nt;
        } catch (Throwable ex) {      // try to cope with OOME
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        nextTable = nextTab;
        transferIndex = n;
    }
    int nextn = nextTab.length;
    // 连接点指针，用于标志位（fwd的hash值为-1，fwd.nextTable=nextTab）
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    // 当advance == true时，表明该节点已经处理过了
    boolean advance = true;
    boolean finishing = false; // to ensure sweep before committing nextTab
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        // 控制 --i ,遍历原hash表中的节点
        while (advance) {
            int nextIndex, nextBound;
            if (--i >= bound || finishing)
                advance = false;
            else if ((nextIndex = transferIndex) <= 0) {

                i = -1;

                advance = false;

            }

            // 用CAS计算得到的transferIndex

            else if (U.compareAndSwapInt
                    (this, TRANSFERINDEX, nextIndex,
                            nextBound = (nextIndex > stride ?
                                    nextIndex - stride : 0))) {
                bound = nextBound;
                i = nextIndex - 1;
                advance = false;
            }
        }

        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            // 已经完成所有节点复制了
            if (finishing) {
                nextTable = null;
                table = nextTab;        // table 指向nextTable
                sizeCtl = (n << 1) - (n >>> 1);     // sizeCtl阈值为原来的1.5倍
                return;     // 跳出死循环，
            }
            // CAS 更扩容阈值，在这里面sizectl值减一，说明新加入一个线程参与到扩容操作
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                    return;
                finishing = advance = true;
                i = n; // recheck before commit
            }
        }
        // 遍历的节点为null，则放入到ForwardingNode 指针节点
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        // f.hash == -1 表示遍历到了ForwardingNode节点，意味着该节点已经处理过了
        // 这里是控制并发扩容的核心
        else if ((fh = f.hash) == MOVED)
            advance = true; // already processed
        else {
            // 节点加锁
            synchronized (f) {
                // 节点复制工作
                if (tabAt(tab, i) == f) {
                    Node<K,V> ln, hn;
                    // fh >= 0 ,表示为链表节点
                    if (fh >= 0) {
                        // 构造两个链表  一个是原链表  另一个是原链表的反序排列
                        int runBit = fh & n;
                        Node<K,V> lastRun = f;
                        for (Node<K,V> p = f.next; p != null; p = p.next) {

                            int b = p.hash & n;

                            if (b != runBit) {

                                runBit = b;

                                lastRun = p;

                            }

                        }

                        if (runBit == 0) {

                            ln = lastRun;

                            hn = null;

                        }

                        else {

                            hn = lastRun;

                            ln = null;

                        }
                        for (Node<K,V> p = f; p != lastRun; p = p.next) {

                            int ph = p.hash; K pk = p.key; V pv = p.val;

                            if ((ph & n) == 0)

                                ln = new Node<K,V>(ph, pk, pv, ln);

                            else

                                hn = new Node<K,V>(ph, pk, pv, hn);
                        }

                        // 在nextTable i 位置处插上链表
                        setTabAt(nextTab, i, ln);
                        // 在nextTable i + n 位置处插上链表
                        setTabAt(nextTab, i + n, hn);
                        // 在table i 位置处插上ForwardingNode 表示该节点已经处理过了
                        setTabAt(tab, i, fwd);
                        // advance = true 可以执行--i动作，遍历节点
                        advance = true;
                    }
                    // 如果是TreeBin，则按照红黑树进行处理，处理逻辑与上面一致
                    else if (f instanceof TreeBin) {
                        TreeBin<K,V> t = (TreeBin<K,V>)f;

                        TreeNode<K,V> lo = null, loTail = null;

                        TreeNode<K,V> hi = null, hiTail = null;

                        int lc = 0, hc = 0;

                        for (Node<K,V> e = t.first; e != null; e = e.next) {

                            int h = e.hash;

                            TreeNode<K,V> p = new TreeNode<K,V>

                                    (h, e.key, e.val, null, null);

                            if ((h & n) == 0) {

                                if ((p.prev = loTail) == null)

                                    lo = p;

                                else

                                    loTail.next = p;

                                loTail = p;

                                ++lc;

                            }

                            else {

                                if ((p.prev = hiTail) == null)

                                    hi = p;

                                else

                                    hiTail.next = p;

                                hiTail = p;

                                ++hc;

                            }

                        }

                        // 扩容后树节点个数若<=6，将树转链表

                        ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :

                                (hc != 0) ? new TreeBin<K,V>(lo) : t;

                        hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :

                                (lc != 0) ? new TreeBin<K,V>(hi) : t;
                        setTabAt(nextTab, i, ln);
                        setTabAt(nextTab, i + n, hn);
                        setTabAt(tab, i, fwd);
                        advance = true;
                    }
                }
            }
        }
    }
}

扩容过程有点复杂，这里主要涉及到多线程并发扩容,ForwardingNode的作用就是支持扩容操作，将已处理的节点和空节点置为ForwardingNode，并发处理时多个线程经过ForwardingNode就表示已经遍历了，就往后遍历，下图是多线程合作扩容的过程：

介绍完扩容过程，我们再次回到put流程，在第四步中是向链表或者红黑树里加节点，到第五步，会调用treeifyBin（）方法进行链表转红黑树的过程。

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
private final void treeifyBin(Node<K,V>[] tab, int index) {

    Node<K,V> b; int n, sc;

    if (tab != null) {

        //如果整个table的数量小于64，就扩容至原来的一倍，不转红黑树了

        //因为这个阈值扩容可以减少hash冲突，不必要去转红黑树

        if ((n = tab.length) < MIN_TREEIFY_CAPACITY) 

            tryPresize(n << 1);

        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {

            synchronized (b) {

                if (tabAt(tab, index) == b) {

                    TreeNode<K,V> hd = null, tl = null;

                    for (Node<K,V> e = b; e != null; e = e.next) {

                        //封装成TreeNode

                        TreeNode<K,V> p =

                            new TreeNode<K,V>(e.hash, e.key, e.val,

                                              null, null);

                        if ((p.prev = tl) == null)

                            hd = p;

                        else

                            tl.next = p;

                        tl = p;

                    }
                    //通过TreeBin对象对TreeNode转换成红黑树
                    setTabAt(tab, index, new TreeBin<K,V>(hd));
                }
            }
        }
    }
}

到第六步表示已经数据加入成功了，现在调用addCount()方法计算ConcurrentHashMap的size，在原来的基础上加一，现在来看看addCount()方法。

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
private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
    //更新baseCount，table的数量，counterCells表示元素个数的变化
    if ((as = counterCells) != null ||
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        CounterCell a; long v; int m;
        boolean uncontended = true;
        //如果多个线程都在执行，则CAS失败，执行fullAddCount，全部加入count
        if (as == null || (m = as.length - 1) < 0 || 
            (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
            !(uncontended =
              U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
            fullAddCount(x, uncontended);
            return;
        }
        if (check <= 1)
            return;
        s = sumCount();
    }
     //check>=0表示需要进行扩容操作
    if (check >= 0) {

        Node<K,V>[] tab, nt; int n, sc;

        while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
               (n = tab.length) < MAXIMUM_CAPACITY) {
            int rs = resizeStamp(n);
            if (sc < 0) {
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||

                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||

                    transferIndex <= 0)

                    break;

                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            //当前线程发起库哦哦让操作，nextTable=null
            else if (U.compareAndSwapInt(this, SIZECTL, sc,

                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}

put的流程现在已经分析完了，你可以从中发现，他在并发处理中使用的是乐观锁，当有冲突的时候才进行并发处理，而且流程步骤很清晰，但是细节设计的很复杂，毕竟多线程的场景也复杂。

get操作

我们现在要回到开始的例子中，我们对个人信息进行了新增之后，我们要获取所新增的信息，使用String name = map.get(“name”)获取新增的name信息，现在我们依旧用debug的方式来分析下ConcurrentHashMap的获取方法get()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    int h = spread(key.hashCode()); //计算两次hash
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {//读取首节点的Node元素
        if ((eh = e.hash) == h) { //如果该节点就是首节点就返回
            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;

        while ((e = e.next) != null) {//既不是首节点也不是ForwardingNode，那就往下遍历
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

ConcurrentHashMap的get操作的流程很简单，也很清晰，可以分为三个步骤来描述：

  1. 计算hash值，定位到该table索引位置，如果是首节点符合就返回1. 如果遇到扩容的时候，会调用标志正在扩容节点ForwardingNode的find方法，查找该节点，匹配就返回1. 以上都不符合的话，就往下遍历节点，匹配就返回，否则最后就返回null
    如果遇到扩容的时候，会调用标志正在扩容节点ForwardingNode的find方法，查找该节点，匹配就返回

size操作

最后我们来看下例子中最后获取size的方式int size = map.size();，现在让我们看下size()方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null)
                sum += a.value;
        }
    }
    return sum;
}

在JDK1.8版本中，对于size的计算，在扩容和addCount()方法就已经有处理了，JDK1.7是在调用size()方法才去计算，其实在并发集合中去计算size是没有多大的意义的，因为size是实时在变的，只能计算某一刻的大小，但是某一刻太快了，人的感知是一个时间段，所以并不是很精确。

总结与思考

其实可以看出JDK1.8版本的ConcurrentHashMap的数据结构已经接近HashMap，相对而言，ConcurrentHashMap只是增加了同步的操作来控制并发，从JDK1.7版本的ReentrantLock+Segment+HashEntry，到JDK1.8版本中synchronized+CAS+HashEntry+红黑树,相对而言，总结如下思考：

  1. JDK1.8的实现降低锁的粒度，JDK1.7版本锁的粒度是基于Segment的，包含多个HashEntry，而JDK1.8锁的粒度就是HashEntry（首节点）

  2. JDK1.8版本的数据结构变得更加简单，使得操作也更加清晰流畅，因为已经使用synchronized来进行同步，所以不需要分段锁的概念，也就不需要Segment这种数据结构了，由于粒度的降低，实现的复杂度也增加了

  3. JDK1.8使用红黑树来优化链表，基于长度很长的链表的遍历是一个很漫长的过程，而红黑树的遍历效率是很快的，代替一定阈值的链表，这样形成一个最佳拍档

  4. JDK1.8为什么使用内置锁synchronized来代替重入锁ReentrantLock，我觉得有以下几点：

    • 因为粒度降低了，在相对而言的低粒度加锁方式，synchronized并不比ReentrantLock差，在粗粒度加锁中ReentrantLock可能通过Condition来控制各个低粒度的边界，更加的灵活，而在低粒度中，Condition的优势就没有了

    • JVM的开发团队从来都没有放弃synchronized，而且基于JVM的synchronized优化空间更大，使用内嵌的关键字比使用API更加自然

    • 在大量的数据操作下，对于JVM的内存压力，基于API的ReentrantLock会开销更多的内存，虽然不是瓶颈，但是也是一个选择依据

参考
https://bentang.me/tech/2016/12/01/jdk8-concurrenthashmap-1/

发表于 分类于 阅读次数： Valine：

简介

服务间调用有三种状态：成功；失败；超时。前两个状态明确，但超时状态完全不知道是什么状态。

阅读全文 »

发表于 分类于 阅读次数： Valine：

常见方式

1、服务的种类来做分离

2、用户维度做分离

服务种类分离

以微服务形式，将业务拆分为用户中心，商品中心，评论中心，各自采用独立的服务器集群部署，独立的数据库。达到物理层面的隔离。

用户维度分离

将用户分成不同的组，并把后端的同一个服务根据这些不同的组分成不同的实例。同一个服务对于不同的用户进行冗余和隔离。这样一个服务实例挂掉，只会影响一部分用户。

“多租户”模式。针对一些大的客户，专用集群。

  • 特点：

增加设计的复杂度，资源上存在浪费。

  • 多租户模式分为三种：

1、完全独立。每个租户有自己完全独立的服务和数据。

2、独立的数据分区，服务共享。（较推荐方式）

3、共享服务，共享数据分区。

设计要点

  • 定义好隔离业务的大小和粒度。过大、过小都不好。需要结合具体业务来看。

  • 考虑系统的复杂度、成本、性能、资源使用的问题。找到合适的均衡方案。

  • 配置一些高可用、重试、异步、消息中间件、流控、熔断等框架

  • 自动化运维工具，象容器或虚拟化技术帮助我们更方便的管理

  • 非常完善的监控系统