asp网站建设实录源码,网页开发软件哪个好用,网站建设维护教程,自己做的网站上传到分析过HashMap的1.7的版本的结构#xff0c;但是HashMap是线程不安全的#xff0c;多线程触发扩容还会发生死循环问题#xff0c;那么ConcurrentHashMap 就是解决这个问题的#xff0c;这是一个线程安全的Map#xff0c;那么对应的内部实现是怎么样的#xff0c;简单分析…分析过HashMap的1.7的版本的结构但是HashMap是线程不安全的多线程触发扩容还会发生死循环问题那么ConcurrentHashMap 就是解决这个问题的这是一个线程安全的Map那么对应的内部实现是怎么样的简单分析下和HashMap相同的位置就不多做重复分析了
构造方法
这是个最基础的构造方法需要的参数有容量扩容因子这是和HashMap相同的地方但是多了一个并发水平选项这里默认值是16也就是并发粒度的控制最多可以16个线程同时加锁对Map处理我们看下具体怎么做的。 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 argumentsint sshift 0;int ssize 1;// 小于并发等级就左移动一位*2倍在得到一个大于等于这个数的2的n次幂while (ssize concurrencyLevel) {sshift;ssize 1;}// sshift 记录的是对应的多少位// ssize记录的就是处理过的并发水平this.segmentShift 32 - sshift;this.segmentMask ssize - 1;if (initialCapacity MAXIMUM_CAPACITY)initialCapacity MAXIMUM_CAPACITY;// 这里在计算分段之后每段的长度int c initialCapacity / ssize;// 这里向上取下因为上面int处理的直接向下取整了if (c * ssize initialCapacity)c;// 最小的分段长度为2int cap MIN_SEGMENT_TABLE_CAPACITY;// 如果小于需要的值就左移动一位扩大二倍知道大于等于需要的while (cap c)cap 1;// create segments and segments[0]// 对应的段段的数组长度就是cap长度的HashEntrySegmentK,V s0 new SegmentK,V(loadFactor, (int)(cap * loadFactor),(HashEntryK,V[])new HashEntry[cap]);// 然后创建段的数组长度就是之前处理过的并发粒度可以保证锁段的时候并发粒度大于等于需要的SegmentK,V[] ss (SegmentK,V[])new Segment[ssize];// cas 赋值UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]// 赋值给Map的成员变量this.segments ss;}从构造方法来看就是对HashMap进行分段了控制实际的容量大于等于需要的容量并发粒度也是大于等于需要的粒度这样可以对每一个段进行加锁保证并发安全又能保证一定的并发粒度后面看下是怎么进行插入和获取的是怎么进行加锁的怎么保证数据安全的
Segment 的结构
看下主要的结构继承了ReentrantLock同样维护了Map的扩容因子扩容阈值元素数量Entry数组这些 static final class SegmentK,V extends ReentrantLock implements Serializable {// 在预扫描之前尝试锁定的最大次数static final int MAX_SCAN_RETRIES Runtime.getRuntime().availableProcessors() 1 ? 64 : 1;// 对应的Entry数组volatile修饰修改数据对其他线程可见及时刷新回主存// transient修饰序列化忽略这个transient volatile HashEntryK,V[] table;// 分段内的元素的数量transient int count;// 修改相关的计数transient int modCount;// 扩容阈值transient int threshold;// 扩容因子final float loadFactor;}put 方法分析
老规矩先从主要方法put开始看put里面一般能看到存储结构查找顺序等关键信息
public V put(K key, V value) {SegmentK,V s;if (value null)throw new NullPointerException();int hash hash(key);// 计算位于哪个segmentint j (hash segmentShift) segmentMask;// segment未初始化化的时候进行初始化if ((s (SegmentK,V)UNSAFE.getObject // nonvolatile; recheck(segments, (j SSHIFT) SBASE)) null) // in ensureSegments ensureSegment(j);// 元素插入return s.put(key, hash, value, false);
}final V put(K key, int hash, V value, boolean onlyIfAbsent) {// 尝试获取锁获取到了第一步获取不到就是后面还是在获取// 存在尝试次数尝试次数完了之后死等如果直接得到锁返回的node就是null// 如果开始没得到锁并且开始数组没数据的时候会得到一个nodeHashEntryK,V node tryLock() ? null :scanAndLockForPut(key, hash, value);// 上一步走完就是获取到锁了V oldValue;try {HashEntryK,V[] tab table;int index (tab.length - 1) hash;// 获取对应下标的值valitale保证及时可见HashEntryK,V first entryAt(tab, index);for (HashEntryK,V e first;;) {// 数组上有值if (e ! null) {K k;// 遇见相同keyif ((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)// 头插法上面获取到了nodenode.setNext(first);elsenode new HashEntryK,V(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;
}put方法就是先检查对应的segment是不是初始化了未初始化的先进性初始化初始化的时候按照第一个进行复制然后cas赋值到对应位置然后执行实际的插入逻辑插入的时候先进行获取lock因为Segnment继承的ReentrantLock直接使用tryLock(),和lock()进行获取的非公平锁获取之后对对应的值进行添加
ensureSegment 初始化segment分析
private SegmentK,V ensureSegment(int k) {final SegmentK,V[] ss this.segments;long u (k SSHIFT) SBASE; // raw offsetSegmentK,V seg;if ((seg (SegmentK,V)UNSAFE.getObjectVolatile(ss, u)) null) {// 用构造方法时创建的第一个segment作为原型复制一个segment出来SegmentK,V proto ss[0]; // use segment 0 as prototypeint cap proto.table.length;float lf proto.loadFactor;int threshold (int)(cap * lf);// 对应的数组HashEntryK,V[] tab (HashEntryK,V[])new HashEntry[cap];if ((seg (SegmentK,V)UNSAFE.getObjectVolatile(ss, u)) null) { // recheckSegmentK,V s new SegmentK,V(lf, threshold, tab);// cas赋值segmentwhile ((seg (SegmentK,V)UNSAFE.getObjectVolatile(ss, u)) null) {if (UNSAFE.compareAndSwapObject(ss, u, null, seg s))break;}}}return seg;
}scanAndLockForPut 这里也是在获取锁能抽空的话就会返回一个HashEntry
private HashEntryK,V scanAndLockForPut(K key, int hash, V value) {// 寻找hash对应的entryHashEntryK,V first entryForHash(this, hash);HashEntryK,V e first;HashEntryK,V node null;int retries -1; // negative while locating node// 获取不到锁就一直获取while (!tryLock()) {HashEntryK,V f; // to recheck first below// 尝试次数if (retries 0) {// 这里的e是遍历的当前节点if (e null) {if (node null) // speculatively create node// 这里的创建Entry应该就是简单的没别的事情做充分利用下node new HashEntryK,V(hash, key, value, null);retries 0;}// 直接遇到相同的了else if (key.equals(e.key))retries 0;// 遍历下一个elsee e.next;}// 超过获取锁的最大尝试次数了else if (retries MAX_SCAN_RETRIES) {lock();break;}// 这里在判断尝试次数是不是为0为0的上面两种一种是对应数组位置无值一种是有相同// key可以直接替换的//然后 后面的 判断了下是不是自己如果不是自己插入的更换下头节点可能是别的线程// 插入的else if ((retries 1) 0 (f entryForHash(this, hash)) ! first) {e first f; // re-traverse if entry changedretries -1;}}return node;
}
static final K,V HashEntryK,V entryForHash(SegmentK,V seg, int h) {HashEntryK,V[] tab;// 判断段是不是空里面数组是不是空不空的时候获取对应的下标return (seg null || (tab seg.table) null) ? null :(HashEntryK,V) UNSAFE.getObjectVolatile(tab, ((long)(((tab.length - 1) h)) TSHIFT) TBASE);
}rehash(node) 扩容方法分析
private void rehash(HashEntryK,V node) {HashEntryK,V[] oldTable table;// 原来的容量int oldCapacity oldTable.length;// 新的容量扩大为2倍int newCapacity oldCapacity 1;threshold (int)(newCapacity * loadFactor);//新的数组HashEntryK,V[] newTable (HashEntryK,V[]) new HashEntry[newCapacity];int sizeMask newCapacity - 1;// 遍历旧的for (int i 0; i oldCapacity ; i) {HashEntryK,V e oldTable[i];if (e ! null) {HashEntryK,V next e.next;int idx e.hash sizeMask;if (next null) // Single node on list// 只有一个newTable[idx] e;else { // Reuse consecutive sequence at same slot// 记录的当前遍历的头节点eHashEntryK,V lastRun e;int lastIdx idx;// 这里在遍历链表last就是当前遍历到的for (HashEntryK,V last next;last ! null;last last.next) {// 计算下标int k last.hash sizeMask;// 这里计算的是新下标的数据if (k ! lastIdx) {// 不在现在这个位置的话lastIdx k;// 这个在记录链表最后一个不在本位置的节点// 获取这个引用的意义就在于后面的不需要转移直接就一串带走lastRun last;}}// 然后赋值给新的位置newTable[lastIdx] lastRun;// Clone remaining nodes// 然后重新遍历了一遍遇到之前lastRun节点停止for (HashEntryK,V p e; p ! lastRun; p p.next) {V v p.value;int h p.hash;int k h sizeMask;// 采用头插法进行遍历插入到对应的链表中HashEntryK,V n newTable[k];newTable[k] new HashEntryK,V(h, p.key, v, n);}}}}// 扩容之后把node插入进去int nodeIndex node.hash sizeMask; // add the new nodenode.setNext(newTable[nodeIndex]);newTable[nodeIndex] node;table newTable;
}可以看到resize基本等同HashMap不过在resize里面把node进行插入的
get 方法
可以看到get方法比较简单不需要加锁通过volitale修饰的key然后UNSAFE.getObjectVolatile 拿到对应的值 public V get(Object key) {SegmentK,V s; // manually integrate access methods to reduce overheadHashEntryK,V[] tab;int h hash(key);long u (((h segmentShift) segmentMask) SSHIFT) SBASE;if ((s (SegmentK,V)UNSAFE.getObjectVolatile(segments, u)) ! null (tab s.table) ! null) {for (HashEntryK,V e (HashEntryK,V) UNSAFE.getObjectVolatile(tab, ((long)(((tab.length - 1) h)) TSHIFT) TBASE);e ! null; e e.next) {K k;if ((k e.key) key || (e.hash h key.equals(k)))return e.value;}}return null;}总结
整体总结下ConcurrentHashMap 通过引入Segment概念来对HashMap进行分段等于有了很多个小的HashMap然后又继承了ReentrantLock通过lock来保证put的数据安全性并发粒度通过设置segment的数组长度来控制默认16可以自定义不过也是2的n次幂类似于原来hashmap的数组大小取值算法对于segment里面数组的最小长度为2这个数组长度和segment的长度决定了容量大小这里会大于等于设置的值segment数组是不可变的也就是map构造完成的时候并发粒度就确定了segment的长度大小不可变扩容是在扩容的内部的HashMap也就是HashEntry数组整体就是hash计算了两次第一次确认在那个segment然后再计算落到那个段里面的哪个位置。
