ConcurrentHashMap和CopyOnWriteArrayList机制

ConcurrentHashMap

介绍：HashMap是线程不安全的，ConcurrentHashMap可以保证线程安全，HashTable也是线程安全的，但是整个容器都是使用synchronized来保证线程安全，效率效率低下；

先介绍一下JDK1.7的ConcurrentHashMap:
分段锁：使用分段锁来保证容器的增删是线程安全的，其原理就是容器中有多把不同的锁，每个锁作用域容器中的一部分数据，那么当多线程访问不同数段的数据时，就不会出现锁竞争，大大降低了操作容器数据时带来的锁竞争；

先看一下源码：
构造函数：

//主要就是创建segments 数组和定义初始数据段的长度
public ConcurrentHashMap(int initialCapacity,
                         float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
        throw new IllegalArgumentException();
    if (concurrencyLevel > MAX_SEGMENTS)
        concurrencyLevel = MAX_SEGMENTS;
    // Find power-of-two sizes best matching arguments
    int sshift = 0;
    int ssize = 1;
//这个是获取Segment[] segments 大小，其大小和HashMap的一样，2的次方幂，并且大于或等于concurrencyLevel，而且最近concurrencyLevel
    while (ssize < concurrencyLevel) {
        ++sshift;
        ssize <<= 1;
    }
//下面都是初始化一下计算定位数组大小的掩码和偏移量，和每段数据的初始化大小
    this.segmentShift = 32 - sshift;
    this.segmentMask = ssize - 1;
    if (initialCapacity > MAXIMUM_CAPACITY)
        initialCapacity = MAXIMUM_CAPACITY;
    int c = initialCapacity / ssize;
    if (c * ssize < initialCapacity)
        ++c;
    int cap = MIN_SEGMENT_TABLE_CAPACITY;
    while (cap < c)
        cap <<= 1;
    // create segments and segments[0]
    Segment s0 =
        new Segment(loadFactor, (int)(cap * loadFactor),
                         (HashEntry[])new HashEntry[cap]);
//创建segments 数组
    Segment[] ss = (Segment[])new Segment[ssize];
    UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
    this.segments = ss;
}

public V put(K key, V value) {
    Segment s;
    if (value == null)
        throw new NullPointerException();
    int hash = hash(key);
    int j = (hash >>> segmentShift) & segmentMask;
    if ((s = (Segment)UNSAFE.getObject          // nonvolatile; recheck
         (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
        s = ensureSegment(j);
    return s.put(key, hash, value, false);
}


put(K key, V value) 方法：

public V put(K key, V value) {
    Segment s;
    if (value == null)
        throw new NullPointerException();
//计算出在segments 数组所在的下标，就是找到其应该存入的数据段
    int hash = hash(key);
    int j = (hash >>> segmentShift) & segmentMask;
    if ((s = (Segment)UNSAFE.getObject          // nonvolatile; recheck
         (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
        s = ensureSegment(j);
//把数据放入这个segment对象中
    return s.put(key, hash, value, false);
}

再看一下存储数据段的Segment对象，继承了ReentrantLock

static final class Segment extends ReentrantLock implements Serializable {
    

    private static final long serialVersionUID = 2249069246763182397L;

    
    static final int MAX_SCAN_RETRIES =
        Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;

    
    transient volatile HashEntry[] table;

    
    transient int count;

    
    transient int modCount;

    
    transient int threshold;

    
    final float loadFactor;

看一下HashEntry，HashEntry[] table的类型，支持单项链表

static final class HashEntry {
    final int hash;
    final K key;
    volatile V value;
    volatile HashEntry next;

接着看put方法：
其就是加锁此方法，其锁对象就是当前的Segment，里面的数据段就是 HashEntry[] table数组，所以不同的Segment的锁是不同的；
其里面每个 table数组数组的扩容也与HashMap类似，说白了每个Segment就是一个小的HashMap，对这个小的HashMap进行增删时，做加锁处理；

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    HashEntry node = tryLock() ? null :
        scanAndLockForPut(key, hash, value);
    V oldValue;
    try {
        HashEntry[] tab = table;
        int index = (tab.length - 1) & hash;
        HashEntry first = entryAt(tab, index);
        for (HashEntry e = first;;) {
            if (e != null) {
                K k;
                if ((k = e.key) == key ||
                    (e.hash == hash && key.equals(k))) {
                    oldValue = e.value;
                    if (!onlyIfAbsent) {
                        e.value = value;
                        ++modCount;
                    }
                    break;
                }
                e = e.next;
            }
            else {
                if (node != null)
                    node.setNext(first);
                else
                    node = new HashEntry(hash, key, value, first);
                int c = count + 1;
                if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                    rehash(node);
                else
                    setEntryAt(tab, index, node);
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        unlock();
    }
    return oldValue;
}

JDK1.8的ConrrentHashMap

存储的数据结构和HashMap的结构是一样的，其保证线程安全的原理是基于三点：
自旋 CAS Synchronized
抛弃了JDK1.7的原来的分段锁；
其大致原理是：
1、如果存储的下标位置等于null，则通过自旋和CAS把元素赋值到指定数组下标
2、如果不为null，则以当前下标的元素作为锁，利用自旋在Synchronized同步代码块中进行元素的追加（和HashMap类似）
3、如果发现容器正在扩容，则进行帮助扩容，转移自己负责的那一部分数据到新数组上
Ps:扩容是的原理是新建一个数组，不影响其查询操作，在完成数据的全部转移之后，在把新数组替换旧数组
table扩容过程就是将table元素迁移到新的table上, 在元素迁移时, 可以并发完成, 加快了迁移速度, 同时不至于阻塞线程。所有元素迁移完成后, 旧的table直接丢失, 直接使用新的table

先看一下其主要的一些属性：

//当Node.hash=MOVED时，代表table正在进行扩容
static final int MOVED     = -1; // hash for forwarding nodes
	
//代表此元素在红黑树
static final int TREEBIN   = -2; // hash for roots of trees
static final int RESERVED  = -3; // hash for transient reservations

//当前元素数组对象
transient volatile Node[] table;


//如果进行扩容，则新建一个数组对象
private transient volatile Node[] nextTable;

//用来标志table的初始化和扩容，
0：代表table还没有初始化
-1：代表正在初始化
小于-1：表明table正在扩容
大于0：初始化完后，table存放的最大元素个数，超过这个就会进行扩容
private transient volatile int sizeCtl;

put(K key, V value)方法：

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();
//计算key的hash值
    int hash = spread(key.hashCode());
    int binCount = 0;
//通过自旋来存入元素
    for (Node[] tab = table;;) {
        Node f; int n, i, fh;
	//如果table素组还没有初始化，则先进行初始化
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
//如果i = (n - 1) & hash)下标的元素为null也就是tab[i]=null，则通过CAS进行赋值
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            if (casTabAt(tab, i, null,
                         new Node(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
//上面的属性已经说了，如果tab[i].hash=MOVED，则代表table正在进行扩容
//当前线程会帮助进行扩容，负责转移一部分数据到新数组上 nextTable 
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
	//以tab[i]元素进行加锁，添加元素到当前下标的桶中；
            synchronized (f) {
	//这一步判断主要是防止并发桶中加锁元素被改变，则要在代码块中判断成功，则这个锁就不会发生改变，因为remove也是使用这个锁
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) {
                        binCount = 1;
                        for (Node 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 pred = e;
                            if ((e = e.next) == null) {
                                pred.next = new Node(hash, key,
                                                          value, null);
                                break;
                            }
                        }
                    }
                    else if (f instanceof TreeBin) {
                        Node p;
                        binCount = 2;
                        if ((p = ((TreeBin)f).putTreeVal(hash, key,
                                                       value)) != null) {
                            oldVal = p.val;
                            if (!onlyIfAbsent)
                                p.val = value;
                        }
                    }
                }
            }
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
// baseCount+1（容器元素加1），检查是否需要进行扩容
    addCount(1L, binCount);
    return null;
}

CopyOnWriteArrayList

核心思想：读写是分离，空间换时间，避免为保证线程安全导致的激烈的锁竞争
（其每次添加一个元素时，都会新建一个数组，再替换老数组）

特点：
1、相对于ArrayList，CopyOnWriteArrayList是线程安全的；
2、适合多线程 ‘读多写少’的场景，因为写是有锁的，读是无锁的；
3、是最终一致性，在写的过程中，原有读的数据不会发生更新，只有新的读才能读到最新的数据；
4、如何使其它的线程读取到最新的数据，要使用volatile进行修饰；
5、在其迭代器上不能使用add、set、remove的操作，否则会抛出UnsupportedOperationException异常

接下来看一下源码：

重要的属性：

//对数据进行更改操作时，加的锁
final transient ReentrantLock lock = new ReentrantLock();


//存放元素的素组
private transient volatile Object[] array;

构造函数：

public CopyOnWriteArrayList() {
    setArray(new Object[0]);
}

final void setArray(Object[] a) {
    array = a;
}

起作用就是Object[] array=new Object[0]；初始化长度为0的数组；

add(E e)方法

public boolean add(E e) {
//一个对象使用的是同一把锁；
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] elements = getArray();
        int len = elements.length;
//每一次插入一个元素时，就会创建一个比原来长度大1的新数组，并把老数组的元素复制到新数组上
        Object[] newElements = Arrays.copyOf(elements, len + 1);
//进行新元素的赋值，添加到新数组的最后一位
        newElements[len] = e;
//把新数组替换到老数组
        setArray(newElements);
        return true;
    } finally {
//解锁
        lock.unlock();
    }
}

get(int index)方法：没有加锁

public E get(int index) {
//拿到上数组对象，获取指定下标的元素
    return get(getArray(), index);
}