株洲网站建设网站运营,如何汉化wordpress主题,西宁市城乡规划建设局网站,南宁制作网站ConcurrentLinkedQueue非阻塞无界链表队列 ConcurrentLinkedQueue是一个线程安全的队列#xff0c;基于链表结构实现#xff0c;是一个无界队列#xff0c;理论上来说队列的长度可以无限扩大。 与其他队列相同#xff0c;ConcurrentLinkedQueue 也采用的是先进先出#… ConcurrentLinkedQueue非阻塞无界链表队列 ConcurrentLinkedQueue是一个线程安全的队列基于链表结构实现是一个无界队列理论上来说队列的长度可以无限扩大。 与其他队列相同ConcurrentLinkedQueue 也采用的是先进先出FIFO入队规则对元素进行排序。当我们向队列中添加元素时新插入的元素会插入到队列的尾部而当我们获取一个元素时它会从队列的头部中取出。 因为 ConcurrentLinkedQueue 是链表结构所以当入队时插入的元素依次向后延伸形成链表而出队时则从链表的第一个元素开始获取依次递增 不知道我这样形容能否让你对链表的入队、出队产生一个大概的思路 ConcurrentLinkedQuere简单示例 值得注意的是在使用ConcurrentLinkedQueue时如果涉及到队列是否为空的判断切记不可使用size()0的做法因为在 size()方法中是通过遍历整个链表来实现的在队列元素很多的时候size()方法十分消耗性能和时间只是单纯的判断队列为空使用isEmpty()即可 public class ConcurrentLinkedQueueTest { public static int threadCount 10; public static ConcurrentLinkedQueueString queue new ConcurrentLinkedQueueString(); static class Offer implements Runnable { public void run() { //不建议使用queue.size()0影响效率。可以使用!queue.isEmpty() if(queue.size()0){ String ele new Random().nextInt(Integer.MAX_VALUE); queue.offer(ele); System.out.println(入队元素为ele); } } } static class Poll implements Runnable { public void run() { if(!queue.isEmpty()){ String ele queue.poll(); System.out.println(出队元素为ele); } } } public static void main(String[] agrs) { ExecutorService executorService Executors.newFixedThreadPool(4); for(int x0;xthreadCount;x){ executorService.submit(new Offer()); executorService.submit(new Poll()); } executorService.shutdown(); } } 一种输出 入队元素为313732926 出队元素为313732926 入队元素为812655435 出队元素为812655435 入队元素为1893079357 出队元素为1893079357 入队元素为1137820958 出队元素为1137820958 入队元素为1965962048 出队元素为1965962048 出队元素为685567162 入队元素为685567162 出队元素为1441081163 入队元素为1441081163 出队元素为1627184732 入队元素为1627184732 ConcurrentLinkedQuere类图 如图 ConcurrentLinkedQueue 中有两个 volatile 类型的 Node 节点分别用来存在列表的首尾节点其中 head节点存放链表第一个 item 为 null 的节点tail 则并不是总指向最后一个节点。Node 节点内部则维护一个变量 item用来存放节点的值next用来存放下一个节点从而链接为一个单向无界列表。
public ConcurrentLinkedQueue() { head tail new NodeE(null);
}
如上代码初始化时候会构建一个item为NULL的空节点作为链表的首尾节点。 ConcurrentLinkedQuere方法
✓ Offer操作
offer操作是在链表末尾添加一个元素下面看看实现原理。
public boolean offer(E e) { //e为null则抛出空指针异常 如图首先查找尾节点qnull,p就是尾节点所以执行p.casNext通过cas设置p的next为新增节点 这时候pt所以不重新设置尾节点为当前新节点。由于多线程可以调用offer方法所以可能两个线程同时执行到了1进行cas那么只有一个会成功假如线程1成功了成功后的链表为 失败的线程会循环一次这时候指针为 这时候会执行3所以pq,然后在循环后指针位置为 所以没有其他线程干扰的情况下会执行1执行cas把新增节点插入到尾部没有干扰的情况下线程2 cas会成 功然后去更新尾节点tail,由于p!t所以更新。这时候链表和指针为 假如线程2cas时候线程3也在执行那么线程3会失败循环一次后线程3的节点状态为 这时候p!t 并且t的原始值为toldt的新值为tnew ,所以told!tnew所以 ptnewtail; 然后在循环一下后节点状态: qnull所以执行1。 现在就差pq这个分支还没有走这个要在执行poll操作后才会出现这个情况。poll后会存在下面的状态 这个时候添加元素时候指针分布为 所以会执行2分支 结果 phead 然后循环循环后指针分布 所以执行(1),然后p!t所以设置tail节点。现在分布图 自引用的节点会被垃圾回收掉。 ✓ add操作 add操作是在链表末尾添加一个元素下面看看实现原理。 其实内部调用的还是offer public boolean add(E e) { return offer(e); } ✓ poll操作 poll操作是在链表头部获取并且移除一个元素下面看看实现原理。 public E poll() { restartFromHead: //死循环 for (;;) { //死循环 for (NodeE h head, p h, q;;) { //保存当前节点值 E item p.item; //当前节点有值则cas变为null1 if (item ! null p.casItem(item, null)) { //cas成功标志当前节点以及从链表中移除 if (p ! h) // 类似tail间隔2设置一次头节点2 updateHead(h, ((q p.next) ! null) ? q : p); return item; } //当前队列为空则返回null3 else if ((q p.next) null) { updateHead(h, p); return null; } //自引用了则重新找新的队列头节点4 else if (p q) continue restartFromHead; else//(5) p q; } } } final void updateHead(NodeE h, NodeE p) { if (h ! p casHead(h, p)) h.lazySetNext(h); } 当队列为空时候 可知执行3这时候有两种情况第一没有其他线程添加元素时候(3)结果为true然后因为h!p为false 所以直接返回null。第二在执行qp.next前其他线程已经添加了一个元素到队列这时候3返回false然后 执行5pq,然后循环后节点分布 这时候执行1分支进行cas把当前节点值值为null同时只有一个线程会成功cas成功 标示该节点 从队列中移除了然后p!h,调用updateHead方法参数为h,p;h!p所以把p变为当前链表head节点然后h节点的next指向自己。现在状态为 cas失败 后 会再次循环这时候分布图为 这时候执行3返回null. 现在还有个分支4没有执行过那么什么时候会执行那 这时候执行1分支进行cas把当前节点值值为null同时只有一个线程A会成功cas成功 标示该 节点从队列中移除了然后p!h,调用updateHead方法假如执行updateHead前另外一个线程B开始poll这时候它p指向为原来的head节点然后当前线程A执行updateHead这时候B线程链表状态为 所以会执行4重新跳到外层循环获取当前head,现在状态为 ✓ peek操作 peek操作是获取链表头部一个元素只读取不移除下面看看实现原理。 代码与poll类似只是少了castItem.并且peek操作会改变head指向offer后head指向哨兵节点第 一次peek后head会指向第一个真的节点元素。 public E peek() { restartFromHead: for (;;) { for (NodeE h head, p h, q;;) { E item p.item; if (item ! null || (q p.next) null) { updateHead(h, p); return item; } else if (p q) continue restartFromHead; else p q; } } } ✓ size操作 获取当前队列元素个数在并发环境下不是很有用因为使用CAS没有加锁所以从调用size函数到返回结 果期间有可能增删元素导致统计的元素个数不精确。 public int size() { int count 0; for (NodeE p first(); p ! null; p succ(p)) if (p.item ! null) // 最大返回Integer.MAX_VALUE if (count Integer.MAX_VALUE) break; return count; } //获取第一个队列元素哨兵元素不算没有则为null NodeE first() { restartFromHead: for (;;) { for (NodeE h head, p h, q;;) { boolean hasItem (p.item ! null); if (hasItem || (q p.next) null) { updateHead(h, p); return hasItem ? p : null; } else if (p q) continue restartFromHead; else p q; } } } //获取当前节点的next元素如果是自引入节点则返回真正头节点 final NodeE succ(NodeE p) { NodeE next p.next; return (p next) ? head : next; } ✓ remove操作 如果队列里面存在该元素则删除给元素如果存在多个则删除第一个并返回true否者返回false public boolean remove(Object o) { //查找元素为空直接返回false if (o null) return false; NodeE pred null; for (NodeE p first(); p ! null; p succ(p)) { E item p.item; //相等则使用cas值null,同时一个线程成功失败的线程循环查找队列中其他元素是否有匹配的。 if (item ! null o.equals(item) p.casItem(item, null)) { //获取next元素 NodeE next succ(p); //如果有前驱节点并且next不为空则链接前驱节点到next, if (pred ! null next ! null) pred.casNext(p, next); return true; } pred p; } return false; } ✓ contains操作 判断队列里面是否含有指定对象由于是遍历整个队列所以类似size 不是那么精确有可能调用该方法 时候元素还在队列里面但是遍历过程中才把该元素删除了那么就会返回false. public boolean contains(Object o) { if (o null) return false; for (NodeE p first(); p ! null; p succ(p)) { E item p.item; if (item ! null o.equals(item)) return true; } return false; } ConcurrentLinkedQuere的offer方法有意思的问题 offer中有个 判断 t ! (t tail假如 tnode1;tailnode2;并且node1!node2那么这个判断是true还 是false那答案是true这个判断是看当前t是不是和tail相等相等则返回true否者为false但是无论 结果是啥执行后t的值都是tail。 下面从字节码来分析下为啥。 • 一个例子 public static void main(String[] args) { int t 2; int tail 3; System.out.println(t ! (t tail)); } 结果为true • 字节码文件 C:\Users\Simple\Desktop\TeacherCode\Crm_Test\build\classes\com\itheima\crm\utiljavap -c Test001 警告: 二进制文件Test001包含com.itheima.crm.util.Test001 Compiled from Test001.java public class com.itheima.crm.util.Test001 { public com.itheima.crm.util.Test001(); Code: 0: aload_0 1: invokespecial #8 // Method java/lang/Object.init:()V 4: return public static void main(java.lang.String[]); Code: 0: iconst_2 1: istore_1 2: iconst_3 3: istore_2 4: getstatic #16 // Field java/lang/System.out:Ljava/io/PrintStream; 7: iload_1 8: iload_2 9: dup 10: istore_1 11: if_icmpeq 18 14: iconst_1 15: goto 19 18: iconst_0 19: invokevirtual #22 // Method java/io/PrintStream.println:(Z)V 22: return } 我们从上面标黄的字节码文件中分析 一开始栈为空 栈 • 第0行指令作用是把值2入栈栈顶元素为2 2 栈 • 第1行指令作用是将栈顶int类型值保存到局部变量t中 栈 • 第2行指令作用是把值3入栈栈顶元素为3 3 栈 • 第3行指令作用是将栈顶int类型值保存到局部变量tail中。 栈 • 第4调用打印命令 • 第7行指令作用是把变量t中的值入栈 2 栈 • 第8行指令作用是把变量tail中的值入栈 3 2 栈 • 现在栈里面的元素为3、2并且3位于栈顶 • 第9行指令作用是当前栈顶元素入栈所以现在栈内容332 3 3 2 栈 • 第10行指令作用是把栈顶元素存放到t现在栈内容32 3 2 栈 • 第 11 行指令作用是判断栈顶两个元素值相等则跳转 18。由于现在栈顶严肃为 32 不相等所以返回true. • 第14行指令作用是把1入栈 • 然后回头分析下!是双目运算符应该是首先把左边的操作数入栈然后在去计算了右侧操作数。 ConcurrentLinkedQuere总结 ConcurrentLinkedQueue使用CAS非阻塞算法实现使用CAS解决了当前节点与next节点之间的安全链接和对当前节点值的赋值。由于使用CAS没有使用锁所以获取size的时候有可能进行offerpoll或者remove操作导致获取的元素个数不精确所以在并发情况下size函数不是很有用。另外第一次peek或者first时候会把head指向第一个真正的队列元素。 下面总结下如何实现线程安全的可知入队出队函数都是操作volatile变量headtail。所以要保证队列线程安全只需要保证对这两个 Node 操作的可见性和原子性由于 volatile 本身保证可见性所以只需要看下多线程下如果保证对着两个变量操作的原子性。 对于offer操作是在tail后面添加元素也就是调用tail.casNext方法而这个方法是使用的CAS操作只有一个线程会成功然后失败的线程会循环一下重新获取tail然后执行casNext方法。对于poll也是这样的。 ➢ ConcurrentHashMap非阻塞Hash集合 ConcurrentHashMap 是 Java 并发包中提供的一个线程安全且高效的 HashMap 实现ConcurrentHashMap在并发编程的场景中使用频率非常之高本文就来分析下ConcurrentHashMap的实现原理并对其实现原理进行分析。 ConcurrentLinkedQuere类图 ConcurrentHashMap 是由 Segment 数组结构和 HashEntry 数组结构组成。Segment 是一种可重入锁ReentrantLock在 ConcurrentHashMap 里扮演锁的角色HashEntry 则用于存储键值对数据。一个ConcurrentHashMap里包含一个Segment数组Segment的结构和HashMap类似是一种数组和链表结构 一个 Segment 里包含一个 HashEntry 数组每个 HashEntry 是一个链表结构的元素 每个 Segment 守护者一个HashEntry数组里的元素,当对HashEntry数组的数据进行修改时必须首先获得它对应的Segment锁。 ConcurrentLinkedQuere实现原理 众所周知哈希表是中非常高效复杂度为 O(1)的数据结构在 Java 开发中我们最常见到最频繁使用的就是HashMap和HashTable但是在线程竞争激烈的并发场景中使用都不够合理。 HashMap 先说HashMapHashMap是线程不安全的在并发环境下可能会形成环状链表扩容时可能造成具体原因自行百度google或查看源码分析导致get操作时cpu空转所以在并发环境中使用HashMap是非常危险的。 HashTable HashTable和HashMap的实现原理几乎一样差别无非是1.HashTable不允许key和value为null2.HashTable 是线程安全的。但是 HashTable 线程安全的策略实现代价却太大了简单粗暴get/put 所有相关操作都是synchronized的这相当于给整个哈希表加了一把大锁多线程访问时候只要有一个线程访问或操作该对象那其他线程只能阻塞相当于将所有的操作串行化在竞争激烈的并发场景中性能就会非常差。如下图 HashTable 性能差主要是由于所有操作需要竞争同一把锁而如果容器中有多把锁每一把锁锁一段数据这样在多线程访问时不同段的数据时就不会存在锁竞争了这样便可以有效地提高并发效率。这就是ConcurrentHashMap所采用的分段锁思想 ConcurrentLinkedQuere源码解析 ConcurrentHashMap采用了非常精妙的分段锁策略ConcurrentHashMap的主干是个Segment数组。 final SegmentK,V[] segments; Segment继承了ReentrantLock所以它就是一种可重入锁ReentrantLock)。在ConcurrentHashMap 一个Segment就是一个子哈希表Segment里维护了一个HashEntry数组并发环境下对于不同Segment 的数据进行操作是不用考虑锁竞争的。就按默认的ConcurrentLeve为16来讲理论上就允许16个线程 并发执行有木有很酷 所以对于同一个Segment的操作才需考虑线程同步不同的Segment则无需考虑。Segment类似于HashMap 一个Segment维护着一个HashEntry数组 transient volatile HashEntryK,V[] table; HashEntry是目前我们提到的最小的逻辑处理单元了。一个ConcurrentHashMap维护一个Segment数组 一个Segment维护一个HashEntry数组。 static final class HashEntryK,V { final int hash; final K key; volatile V value; volatile HashEntryK,V next; //其他省略 } 我们说 Segment 类似哈希表那么一些属性就跟我们之前提到的 HashMap 差不离比如负载因子 loadFactor比如阈值threshold等等看下Segment的构造方法 Segment(float lf, int threshold, HashEntryK,V[] tab) { this.loadFactor lf;//负载因子 this.threshold threshold;//阈值 this.table tab;//主干数组即HashEntry数组 } 我们来看下ConcurrentHashMap的构造方法 public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { if (!(loadFactor 0) || initialCapacity 0 || concurrencyLevel 0) throw new IllegalArgumentException(); //MAX_SEGMENTS 为11665536也就是最大并发数为65536 if (concurrencyLevel MAX_SEGMENTS) concurrencyLevel MAX_SEGMENTS; //2的sshif次方等于ssize例:ssize16,sshift4;ssize32,sshif5 int sshift 0; //ssize 为segments数组长度根据concurrentLevel计算得出 int ssize 1; while (ssize concurrencyLevel) { sshift; ssize 1; } //segmentShift和segmentMask这两个变量在定位segment时会用到后面会详细讲 this.segmentShift 32 - sshift; this.segmentMask ssize - 1; if (initialCapacity MAXIMUM_CAPACITY) initialCapacity MAXIMUM_CAPACITY; //计算cap的大小即Segment中HashEntry的数组长度cap也一定为2的n次方. int c initialCapacity / ssize; if (c * ssize initialCapacity) c; int cap MIN_SEGMENT_TABLE_CAPACITY; while (cap c) cap 1; //创建segments数组并初始化第一个Segment其余的Segment延迟初始化 SegmentK,V s0 new SegmentK,V(loadFactor, (int)(cap * loadFactor), (HashEntryK,V[])new HashEntry[cap]); SegmentK,V[] ss (SegmentK,V[])new Segment[ssize]; UNSAFE.putOrderedObject(ss, SBASE, s0); this.segments ss; } 初始化方法有三个参数如果用户不指定则会使用默认值initialCapacity 为 16loadFactor 为 0.75负载因 子扩容时需要参考concurrentLevel为16。 从上面的代码可以看出来,Segment数组的大小ssize是由concurrentLevel来决定的但是却不一定等于 concurrentLevelssize 一定是大于或等于 concurrentLevel 的最小的 2 的次幂。比如默认情况下 concurrentLevel 是 16则 ssize为 16若concurrentLevel 为14ssize为16若 concurrentLevel为 17则ssize为32。为什么Segment的数组大小一定是2的次幂其实主要是便于通过按位与的散列算法来定 位Segment的index。 其实put方法对segment也会有所体现 public V put(K key, V value) { SegmentK,V s; //concurrentHashMap不允许key/value为空 if (value null) throw new NullPointerException(); //hash函数对key的hashCode重新散列避免差劲的不合理的hashcode保证散列均匀 int hash hash(key); //返回的hash值无符号右移segmentShift位与段掩码进行位运算定位segment int j (hash segmentShift) segmentMask; if ((s (SegmentK,V)UNSAFE.getObject // nonvolatile; recheck (segments, (j SSHIFT) SBASE)) null) // in ensureSegment s ensureSegment(j); return s.put(key, hash, value, false); } 从源码看出put的主要逻辑也就两步 1.定位segment并确保定位的Segment已初始化 2.调用Segment的put方法。 Ps关于segmentShift和segmentMask segmentShift和segmentMask这两个全局变量的主要作用是用来定位Segmentint j (hash segmentShift) segmentMask。 segmentMask段掩码假如 segments 数组长度为 16则段掩码为 16-115segments 长度为32段掩码为32-131。这样得到的所有bit位都为1可以更好地保证散列的均匀性 segmentShift2 的 sshift 次方等于 ssizesegmentShift32-sshift。若 segments 长度为 16segmentShift32-428;若 segments 长度为 32segmentShift32-527。而计算得出的 hash 值最大为32位无符号右移segmentShift则意味着只保留高几位其余位是没用的然后与段掩码segmentMask位运算来定位Segment。 ConcurrentLinkedQuere方法 ✓ Get操作 public V get(Object key) { SegmentK,V s; HashEntryK,V[] tab; int h hash(key); long u (((h segmentShift) segmentMask) SSHIFT) SBASE; //先定位Segment再定位HashEntry 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; } get 方法无需加锁由于其中涉及到的共享变量都使用 volatile 修饰volatile 可以保证内存可见性所以 不会读取到过期数据。 来看下 concurrentHashMap 代理到 Segment 上的 put 方法Segment 中的 put 方法是要加锁的。 只不过是锁粒度细了而已。 ✓ Put操作 final V put(K key, int hash, V value, boolean onlyIfAbsent) { HashEntryK,V node tryLock() ? null : scanAndLockForPut(key, hash, value);//tryLock 不成功时会遍历定位到的 HashEnry 位置的链表 遍历主要是为了使 CPU 缓存链表若找不到则创建 HashEntry。tryLock 一定次数后MAX_SCAN_RETRIES 变量决定则lock。若遍历过程中由于其他线程的操作导致链表头结点变化则需要重新遍历。 V oldValue; try { HashEntryK,V[] tab table; int index (tab.length - 1) hash;//定位HashEntry可以看到这个hash值在定位Segment 时和在Segment中定位HashEntry都会用到只不过定位Segment时只用到高几位。 HashEntryK,V first entryAt(tab, index); for (HashEntryK,V 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 HashEntryK,V(hash, key, value, first); int c count 1; //若c超出阈值threshold需要扩容并rehash。扩容后的容量是当前容量的2倍。这样可以最大程 度避免之前散列好的entry重新散列具体在另一篇文章中有详细分析不赘述。扩容并rehash的这个过程是比较消耗资源的。 if (c threshold tab.length MAXIMUM_CAPACITY) rehash(node); else setEntryAt(tab, index, node); modCount; count c; oldValue null; break; } } } finally { unlock(); } return oldValue; } ConcurrentLinkedQuere总结 ConcurrentHashMap 作为一种线程安全且高效的哈希表的解决方案尤其其中的分段锁的方案相比 HashTable的全表锁在性能上的提升非常之大。本文对ConcurrentHashMap的实现原理进行了详细分析并解读了 部分源码希望能帮助到有需要的童鞋。 ➢ ConcurrentSkipListMap非阻塞Hash跳表集合 大家都是知道 TreeMap它是使用树形结构的方式进行存储数据的线程不安全的 Map 集合有序的哈希表 并且可以对Map中的Key进行排序Key中存储的数据需要实现Comparator接口或使用CompareAble接口的子类来实现排序。 ConcurrentSkipListMap也是和TreeMap它们都是有序的哈希表。但是它们是有区别的 第一它们的线程安全机制不同TreeMap 是非线程安全的而 ConcurrentSkipListMap 是线程安全的。第二ConcurrentSkipListMap是通过跳表实现的而TreeMap是通过红黑树实现的。 那现在我们需要知道说明是跳表。 什么是SkipList Skip list(跳表是一种可以代替平衡树的数据结构默认是按照Key值升序的。Skip list让已排序的数据分布在多层链表中以 0-1 随机数决定一个数据的向上攀升与否通过“空间来换取时间”的一个算法在每个节点中增加了向前的指针在插入、删除、查找时可以忽略一些不可能涉及到的结点从而提高了效率。 从概率上保持数据结构的平衡比显示的保持数据结构平衡要简单的多。对于大多数应用用Skip list要比用树算法相对简单。由于Skip list比较简单实现起来会比较容易虽然和平衡树有着相同的时间复杂度(O(logn)),但是skip list的常数项会相对小很多。Skip list在空间上也比较节省。一个节点平均只需要1.333个指针甚至更少。 下图为Skip list结构图以7,14,21,32,37,71,85序列为例 SkipList性质 (1) 由很多层结构组成level是通过一定的概率随机产生的。 (2) 每一层都是一个有序的链表默认是升序也可以根据创建映射时所提供的 Comparator 进行排序具体取决于使用的构造方法。 (3) 最底层(Level 1)的链表包含所有元素。 (4) 如果一个元素出现在Level i 的链表中则它在Level i 之下的链表也都会出现。 (5) 每个节点包含两个指针一个指向同一链表中的下一个元素一个指向下面一层的元素。 什么是ConcurrentSkipListMap ConcurrentSkipListMap提供了一种线程安全的并发访问的排序映射表。内部是SkipList跳表结构实现在理论上能够在O(log(n))时间内完成查找、插入、删除操作。 注意调用ConcurrentSkipListMap的size时由于多个线程可以同时对映射表进行操作所以映射表需要遍历整个链表才能返回元素个数这个操作是个O(log(n))的操作。 ConcurrentSkipListMap数据结构 ConcurrentSkipListMap的数据结构如下图所示 说明可以看到 ConcurrentSkipListMap 的数据结构使用的是跳表每一个 HeadIndex、Index 结点都会包含一个对Node的引用同一垂直方向上的Index、HeadIndex结点都包含了最底层的Node结点的引用。并且层级越高该层级的结点HeadIndex和Index数越少。Node结点之间使用单链表结构。 ConcurrentSkipListMap源码分析 ConcurrentSkipListMap主要用到了Node和Index两种节点的存储方式通过volatile关键字实现了并发的操作 static final class NodeK,V { final K key; volatile Object value;//value值 volatile NodeK,V next;//next引用 …… } static class IndexK,V { final NodeK,V node; final IndexK,V down;//downy引用 volatile IndexK,V right;//右边引用 …… } ✓ ConcurrentSkipListMap查找操作 通过SkipList的方式进行查找操作下图以“查找91”进行说明 红色虚线表示查找的路径蓝色向右箭头表示right引用黑色向下箭头表示down引用 下面就是源码中的实现get方法是通过doGet方法来时实现的 public V get(Object key) { return doGet(key); } //doGet的实现 private V doGet(Object okey) { Comparable? super K key comparable(okey); NodeK,V bound null; IndexK,V q head;//把头结点作为当前节点的前驱节点 IndexK,V r q.right;//前驱节点的右节点作为当前节点 NodeK,V n; K k; int c; for (;;) {//遍历 IndexK,V d; // 依次遍历right节点 if (r ! null (n r.node) ! bound (k n.key) ! null) { if ((c key.compareTo(k)) 0) {//由于key都是升序排列的所有当前关键字大于所要 查找的key时继续向右遍历 q r; r r.right; continue; } else if (c 0) { //如果找到了相等的key节点则返回该Node的value如果value为空可能是其他并发delete 导致的于是通过另一种 //遍历findNode的方式再查找 Object v n.value; return (v ! null)? (V)v : getUsingFindNode(key); } else bound n; } //如果一个链表中right没能找到key对应的value则调整到其down的引用处继续查找 if ((d q.down) ! null) { q d; r d.right; } else break; } // 如果通过上面的遍历方式还没能找到key对应的value再通过Node.next的方式进行查找 for (n q.node.next; n ! null; n n.next) { if ((k n.key) ! null) { if ((c key.compareTo(k)) 0) { Object v n.value; return (v ! null)? (V)v : getUsingFindNode(key); } else if (c 0) break; } } return null; } ✓ ConcurrentSkipListMap删除操作 通过SkipList的方式进行删除操作下图以“删除23”进行说明 红色虚线表示查找的路径蓝色向右箭头表示right引用黑色向下箭头表示down引用 下面就是源码中的实现remove方法是通过doRemove方法来时实现的 //remove操作通过doRemove实现把所有level中出现关键字key的地方都delete掉 public V remove(Object key) { return doRemove(key, null); } final V doRemove(Object okey, Object value) { Comparable? super K key comparable(okey); for (;;) { NodeK,V b findPredecessor(key);//得到key的前驱就是比key小的最大节点 NodeK,V n b.next;//前驱节点的next引用 for (;;) {//遍历 if (n null)//如果next引用为空直接返回 return null; NodeK,V f n.next; if (n ! b.next) // 如果两次获得的b.next不是相同的Node就跳转 到第一层循环重新获得b和n break; Object v n.value; if (v null) { // 当 n 被其他线程 delete 的时候其 valuenull 此时做辅助处理并重新获取b和n n.helpDelete(b, f); break; } if (v n || b.value null) // 当其前驱被delet的时候直接跳出重新获取b和n break; int c key.compareTo(n.key); if (c 0) return null; if (c 0) {//当key较大时就继续遍历 b n; n f; continue; } if (value ! null !value.equals(v)) return null; if (!n.casValue(v, null)) break; if (!n.appendMarker(f) || !b.casNext(n, f))//casNext方法就是通过比较和设置b前驱 的next节点的方式来实现删除操作 findNode(key); // 通过尝试findNode的方式继续find else { findPredecessor(key); // Clean index if (head.right null) //如果head的right引用为空则表示不存在该level tryReduceLevel(); } return (V)v; } } } ✓ ConcurrentSkipListMap插入操作 通过SkipList的方式进行插入操作下图以“添加55”的两种情况进行说明 在level2该level存在的情况下添加55的图示:只需在level2的合适位置插入55即可 在 level4(该 level 不存在图示 level4 是新建的)的情况下添加55 的情况首先新建 level4,然后在level4 的合适位置插入55。 下面就是源码中的实现put方法是通过doPut方法来时实现的 //put操作通过doPut实现 public V put(K key, V value) { if (value null) throw new NullPointerException(); return doPut(key, value, false); } private V doPut(K kkey, V value, boolean onlyIfAbsent) { Comparable? super K key comparable(kkey); for (;;) { NodeK,V b findPredecessor(key);//前驱 NodeK,V n b.next; //定位的过程就是和get操作相似 for (;;) { if (n ! null) { NodeK,V f n.next; if (n ! b.next) // 前后值不一致的情况下跳转到第一层循环重新获得b和n break;; Object v n.value; if (v null) { // n被delete的情况下 n.helpDelete(b, f); break; } if (v n || b.value null) // b 被delete的情况重新获取b和n break; int c key.compareTo(n.key); if (c 0) { b n; n f; continue; } if (c 0) { if (onlyIfAbsent || n.casValue(v, value)) return (V)v; else break; // restart if lost race to replace value } // else c 0; fall through } NodeK,V z new NodeK,V(kkey, value, n); if (!b.casNext(n, z)) break; // restart if lost race to append to b int level randomLevel();//得到一个随机的level作为该key-value插入的最高level if (level 0) insertIndex(z, level);//进行插入操作 return null; } } } /** * 获得一个随机的level值 */ private int randomLevel() { int x randomSeed; x ^ x 13; x ^ x 17; randomSeed x ^ x 5; if ((x 0x8001) ! 0) // test highest and lowest bits return 0; int level 1; while (((x 1) 1) ! 0) level; return level; } //执行插入操作如上图所示有两种可能的情况 //1.当level存在时对leveln都执行insert操作 //2.当level不存在大于目前的最大level时首先添加新的level然后在执行操作1 private void insertIndex(NodeK,V z, int level) { HeadIndexK,V h head; int max h.level; if (level max) {//情况1 IndexK,V idx null; for (int i 1; i level; i)//首先得到一个包含1~level个级别的down关系的链表 最后的inx为最高level idx new IndexK,V(z, idx, null); addIndex(idx, h, level);//把最高level的idx传给addIndex方法 } else { // 情况2 增加一个新的级别 level max 1; IndexK,V[] idxs (IndexK,V[])new Index[level1]; IndexK,V idx null; for (int i 1; i level; i)//该步骤和情况1类似 idxs[i] idx new IndexK,V(z, idx, null); HeadIndexK,V oldh; int k; for (;;) { oldh head; int oldLevel oldh.level; if (level oldLevel) { // lost race to add level k level; break; } HeadIndexK,V newh oldh; NodeK,V oldbase oldh.node; for (int j oldLevel1; j level; j) newh new HeadIndexK,V(oldbase, newh, idxs[j], j);//创建新的 if (casHead(oldh, newh)) { k oldLevel; break; } } addIndex(idxs[k], oldh, k); } } /** *在1~indexlevel层中插入数据 */ private void addIndex(IndexK,V idx, HeadIndexK,V h, int indexLevel) { // insertionLevel 代表要插入的level该值会在indexLevel~1间遍历一遍 int insertionLevel indexLevel; Comparable? super K key comparable(idx.node.key); if (key null) throw new NullPointerException(); // 和get操作类似不同的就是查找的同时在各个level上加入了对应的key for (;;) { int j h.level; IndexK,V q h; IndexK,V r q.right; IndexK,V t idx; for (;;) { if (r ! null) { NodeK,V n r.node; // compare before deletion check avoids needing recheck int c key.compareTo(n.key); if (n.value null) { if (!q.unlink(r)) break; r q.right; continue; } if (c 0) { q r; r r.right; continue; } } if (j insertionLevel) {//在该层level中执行插入操作 // Dont insert index if node already deleted if (t.indexesDeletedNode()) { findNode(key); // cleans up return; } if (!q.link(r, t))//执行link操作其实就是inset的实现部分 break; // restart if (--insertionLevel 0) { // need final deletion check before return if (t.indexesDeletedNode()) findNode(key); return; } } if (--j insertionLevel j indexLevel)//key移动到下一层level t t.down; q q.down; r q.right; } } } ConcurrentLinkedQuere示例 下面我们看下面示例输出的结果 import java.util.*;import java.util.concurrent.*; /* * ConcurrentSkipListMap是“线程安全”的哈希表而TreeMap是非线程安全的。 * * 下面是“多个线程同时操作并且遍历map”的示例 * (01) 当map是ConcurrentSkipListMap对象时程序能正常运行。 * (02) 当map是TreeMap对象时程序会产生ConcurrentModificationException异常。 * */public class ConcurrentSkipListMapDemo1 { // TODO: map是TreeMap对象时程序会出错。 //private static MapString, String map new TreeMapString, String(); private static MapString, String map new ConcurrentSkipListMapString, String(); public static void main(String[] args) { // 同时启动两个线程对map进行操作 new MyThread(a).start(); new MyThread(b).start(); } private static void printAll() { String key, value; Iterator iter map.entrySet().iterator(); while(iter.hasNext()) { Map.Entry entry (Map.Entry)iter.next(); key (String)entry.getKey(); value (String)entry.getValue(); System.out.print((key, value), ); } System.out.println(); } private static class MyThread extends Thread { MyThread(String name) { super(name); } Override public void run() { int i 0; while (i 6) { // “线程名” 序号 String val Thread.currentThread().getName()i; map.put(val, 0); // 通过“Iterator”遍历map。 printAll(); } } } } 某一次的运行结果 (a1, 0), (a1, 0), (b1, 0), (b1, 0), (a1, 0), (b1, 0), (b2, 0), (a1, 0), (a1, 0), (a2, 0), (a2, 0), (b1, 0), (b1, 0), (b2, 0), (b2, 0), (b3, 0), (b3, 0), (a1, 0), (a2, 0), (a3, 0), (a1, 0), (b1, 0), (a2, 0), (b2, 0), (a3, 0), (b3, 0), (b1, 0), (b4, 0), (b2, 0), (a1, 0), (b3, 0), (a2, 0), (b4, 0), (a3, 0), (a1, 0), (a4, 0), (a2, 0), (b1, 0), (a3, 0), (b2, 0), (a4, 0), (b3, 0), (b1, 0), (b4, 0), (b2, 0), (b5, 0), (b3, 0), (a1, 0), (b4, 0), (a2, 0), (b5, 0), (a3, 0), (a1, 0), (a4, 0), (a2, 0), (a5, 0), (a3, 0), (b1, 0), (a4, 0), (b2, 0), (a5, 0), (b3, 0), (b1, 0), (b4, 0), (b2, 0), (b5, 0), (b3, 0), (b6, 0), (b4, 0), (a1, 0), (b5, 0), (a2, 0), (b6, 0), (a3, 0), (a4, 0), (a5, 0), (a6, 0), (b1, 0), (b2, 0), (b3, 0), (b4, 0), (b5, 0), (b6, 0), 结果说明 示例程序中启动两个线程(线程a和线程b)分别对ConcurrentSkipListMap进行操作。以线程a而言它会先获取“线程名”“序号”然后将该字符串作为key将“0”作为value插入到ConcurrentSkipListMap中接着遍历并输出ConcurrentSkipListMap中的全部元素。 线程b的操作和线程a一样只不过线程b的名字和线程a的名字不同。 当 map 是 ConcurrentSkipListMap 对象时程序能正常运行。如果将 map 改为 TreeMap 时程序会产生ConcurrentModificationException异常。 2) java.util.concurrent.atomic包 ➢ AtomicBoolean原子性布尔 AtomicBoolean是java.util.concurrent.atomic包下的原子变量这个包里面提供了一组原子类。其基本的特性就是在多线程环境下当有多个线程同时执行这些类的实例包含的方法时具有排他性即当某个线程进入方法执行其中的指令时不会被其他线程打断而别的线程就像自旋锁一样一直等到该方法执行完成才由JVM从等待队列中选择一个另一个线程进入这只是一种逻辑上的理解。实际上是借助硬件的相关指令来实现的不会阻塞线程(或者说只是在硬件级别上阻塞了)。 AtomicBoolean在这个Boolean值的变化的时候不允许在之间插入保持操作的原子性。 下面将解释重点方法并举例 boolean compareAndSet(expectedValue, updateValue)这个方法主要两个作用: 1. 比较AtomicBoolean和expect的值如果一致执行方法内的语句。其实就是一个if语句 2. 把 AtomicBoolean 的值设成 update,比较最要的是这两件事是一气呵成的这连个动作之间不会被 打断任何内部或者外部的语句都不可能在两个动作之间运行。为多线程的控制提供了解决的方案 下面我们从代码上解释 首先我们看下在不使用AtomicBoolean情况下代码的运行情况 package zmx.atomic.test; import java.util.concurrent.TimeUnit; public class BarWorker implements Runnable { //静态变量 private static boolean exists false; private String name; public BarWorker(String name) { this.name name; } Override public void run() { if (!exists) { try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e1) { // do nothing } exists true; System.out.println(name enter); try { System.out.println(name working); TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { // do nothing } System.out.println(name leave); exists false; } else { System.out.println(name give up); } } public static void main(String[] args) { BarWorker bar1 new BarWorker(bar1); BarWorker bar2 new BarWorker(bar2); new Thread(bar1).start(); new Thread(bar2).start(); } } 运行结果 bar1 enter bar2 enter bar1 working bar2 working bar1 leave bar2 leave 从上面的运行结果我们可看到两个线程运行时都对静态变量exists同时做操作并没有保证exists静态变量 的原子性也就是一个线程在对静态变量 exists 进行操作到时候其他线程必须等待或不作为。等待一个线程操作完 后才能对其进行操作。 下面我们将静态变量使用AtomicBoolean来进行操作 package zmx.atomic.test; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; public class BarWorker2 implements Runnable { //静态变量使用AtomicBoolean 进行操作 private static AtomicBoolean exists new AtomicBoolean(false); private String name; public BarWorker2(String name) { this.name name; } Override public void run() { if (exists.compareAndSet(false, true)) { System.out.println(name enter); try { System.out.println(name working); TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { // do nothing } System.out.println(name leave); exists.set(false); } else { System.out.println(name give up); } } public static void main(String[] args) { BarWorker2 bar1 new BarWorker2(bar1); BarWorker2 bar2 new BarWorker2(bar2); new Thread(bar1).start(); new Thread(bar2).start(); } } 运行结果 bar1 enter bar1 working bar2 give up bar1 leave 可以从上面的运行结果看出仅仅一个线程进行工作因为exists.compareAndSet(false, true)提供了原子性操作 比较和赋值操作组成了一个原子操作, 中间不会提供可乘之机。使得一个线程操作其他线程等待或不作为。 下面我们简单介绍下AtomicBoolean的API 创建一个 AtomicBoolean 你可以这样创建一个 AtomicBoolean AtomicBoolean atomicBoolean new AtomicBoolean(); 以上示例新建了一个默认值为 false 的 AtomicBoolean。如果你想要为 AtomicBoolean 实例设置一个显式的 初始值那么你可以将初始值传给 AtomicBoolean 的构造子 AtomicBoolean atomicBoolean new AtomicBoolean(true); 获得 AtomicBoolean的值 你可以通过使用 get() 方法来获取一个 AtomicBoolean 的值。示例如下 AtomicBoolean atomicBoolean new AtomicBoolean(true); boolean value atomicBoolean.get(); 设置AtomicBoolean的值 你可以通过使用 set() 方法来设置一个 AtomicBoolean 的值。 示例如下 AtomicBoolean atomicBoolean new AtomicBoolean(true); atomicBoolean.set(false); 以上代码执行后 AtomicBoolean 的值为 false。 交换AtomicBoolean的值 你可以通过 getAndSet() 方法来交换一个 AtomicBoolean 实例的值。getAndSet() 方法将返回 AtomicBoolean 当前的值并将为 AtomicBoolean 设置一个新值。示例如下 AtomicBoolean atomicBoolean new AtomicBoolean(true); boolean oldValue atomicBoolean.getAndSet(false); 以上代码执行后 oldValue 变量的值为 trueatomicBoolean 实例将持有 false 值。代码成功将 AtomicBoolean 当前值 ture 交换为 false。 比较并设置AtomicBoolean的值 compareAndSet() 方法允许你对 AtomicBoolean 的当前值与一个期望值进行比较如果当前值等于期望值的话将会对 AtomicBoolean 设定一个新值。compareAndSet() 方法是原子性的因此在同一时间之内有单个线程执行它。因此 compareAndSet() 方法可被用于一些类似于锁的同步的简单实现。以下是一个 compareAndSet() 示例 AtomicBoolean atomicBoolean new AtomicBoolean(true); boolean expectedValue true; boolean newValue false; boolean wasNewValueSet atomicBoolean.compareAndSet( expectedValue, newValue); 本示例对 AtomicBoolean 的当前值与 true 值进行比较如果相等将 AtomicBoolean 的值更新为 false ➢ AtomicInteger原子性整型 AtomicInteger一个提供原子操作的Integer的类。在Java语言中i和i操作并不是线程安全的 在使用的时候不可避免的会用到 synchronized关键字。而AtomicInteger 则通过一种线程安全的加减操 作接口。 我们先来看看AtomicInteger给我们提供了什么方法 public final int get() //获取当前的值 public final int getAndSet(int newValue)//获取当前的值并设置新的值 public final int getAndIncrement()//获取当前的值并自增 public final int getAndDecrement() //获取当前的值并自减 public final int getAndAdd(int delta) //获取当前的值并加上预期的值 下面通过两个简单的例子来看一下 AtomicInteger 的优势在哪: 普通线程同步: class Test2 { private volatile int count 0; public synchronized void increment() { count; //若要线程安全执行执行count需要加锁 } public int getCount() { return count; } } 使用AtomicInteger: class Test2 { private AtomicInteger count new AtomicInteger(); public void increment() { count.incrementAndGet(); } //使用AtomicInteger之后不需要加锁也可以实现线程安全。 public int getCount() { return count.get(); } } 从上面的例子中我们可以看出使用 AtomicInteger 是非常的安全的.而且因为 AtomicInteger 由硬件提供原子 操作指令实现的。在非激烈竞争的情况下开销更小速度更快。AtomicInteger 是使用非阻塞算法来实现并发控制 的。AtomicInteger的关键域只有一下3个 // setup to use Unsafe.compareAndSwapInt for updates private static final Unsafe unsafe Unsafe.getUnsafe(); private static final long valueOffset; static { try { valueOffset unsafe.objectFieldOffset(AtomicInteger.class.getDeclaredField(value)); } catch (Exception ex) { throw new Error(ex); } } private volatile int value; 这里 unsafe是java提供的获得对对象内存地址访问的类注释已经清楚的写出了它的作用就是在更新操作 时提供“比较并替换”的作用。实际上就是 AtomicInteger 中的一个工具。valueOffset 是用来记录 value 本身在内 存的便宜地址的这个记录也主要是为了在更新操作在内存中找到value的位置方便比较。 注意value是用来存储整数的时间变量这里被声明为volatile就是为了保证在更新操作时当前线程可以拿 到value最新的值并发环境下value可能已经被其他线程更新了。 优点:最大的好处就是可以避免多线程的优先级倒置和死锁情况的发生提升在高并发处理下的性能。 下面我们简单介绍下AtomicInteger的API 创建一个AtomicInteger 创建一个 AtomicInteger 示例如下 AtomicInteger atomicInteger new AtomicInteger(); 本示例将创建一个初始值为 0 的 AtomicInteger。如果你想要创建一个给定初始值的 AtomicInteger你可 以这样 AtomicInteger atomicInteger new AtomicInteger(123); 本示例将 123 作为参数传给 AtomicInteger 的构造子它将设置 AtomicInteger 实例的初始值为 123。 获得AtomicInteger的值 你可以使用 get() 方法获取 AtomicInteger 实例的值。示例如下 AtomicInteger atomicInteger new AtomicInteger(123); int theValue atomicInteger.get(); 设置AtomicInteger的值 你可以通过 set() 方法对 AtomicInteger 的值进行重新设置。以下是 AtomicInteger.set() 示例 AtomicInteger atomicInteger new AtomicInteger(123); atomicInteger.set(234); 以上示例创建了一个初始值为 123 的 AtomicInteger而在第二行将其值更新为 234。 比较并设置AtomicInteger的值 AtomicInteger 类也通过了一个原子性的 compareAndSet() 方法。这一方法将 AtomicInteger 实例的当前值 与期望值进行比较如果二者相等为 AtomicInteger 实例设置一个新值。AtomicInteger.compareAndSet() 代码 示例 AtomicInteger atomicInteger new AtomicInteger(123); int expectedValue 123; int newValue 234; atomicInteger.compareAndSet(expectedValue, newValue); 本示例首先新建一个初始值为 123 的 AtomicInteger 实例。然后将 AtomicInteger 与期望值 123 进行比较 如果相等将 AtomicInteger 的值更新为 234。 增加AtomicInteger的值 AtomicInteger 类包含有一些方法通过它们你可以增加 AtomicInteger 的值并获取其值。这些方法如下 public final int addAndGet(int addValue)//在原来的数值上增加新的值并返回新值 public final int getAndIncrement()//获取当前的值并自增 public final int incrementAndget() //自减并获得自减后的值 public final int getAndAdd(int delta) //获取当前的值并加上预期的值 第一个 addAndGet() 方法给 AtomicInteger 增加了一个值然后返回增加后的值。getAndAdd() 方法为 AtomicInteger 增加了一个值但返回的是增加以前的 AtomicInteger 的值。具体使用哪一个取决于你的应用场景。 以下是这两种方法的示例 AtomicInteger atomicInteger new AtomicInteger(); System.out.println(atomicInteger.getAndAdd(10)); System.out.println(atomicInteger.addAndGet(10)); 本示例将打印出 0 和 20。例子中第二行拿到的是加 10 之前的 AtomicInteger 的值。加 10 之前的值是 0。第三行将 AtomicInteger 的值再加 10并返回加操作之后的值。该值现在是为 20。你当然也可以使用这俩方法为 AtomicInteger 添加负值。结果实际是一个减法操作。getAndIncrement() 和 incrementAndGet() 方法类似于 getAndAdd() 和 addAndGet()但每次只将 AtomicInteger 的值加 1。 减小AtomicInteger的值 AtomicInteger 类还提供了一些减小 AtomicInteger 的值的原子性方法。这些方法是 public final int decrementAndGet() public final int getAndDecrement() decrementAndGet() 将 AtomicInteger 的值减一并返回减一后的值。getAndDecrement() 也将 AtomicInteger 的值减一但它返回的是减一之前的值。 ➢ AtomicIntegerArray原子性整型数组 java.util.concurrent.atomic.AtomicIntegerArray 类提供了可以以原子方式读取和写入的底层 int 数组的操作 还包含高级原子操作。 AtomicIntegerArray 支持对底层 int 数组变量的原子操作。 它具有获取和设置方法如在变 量上的读取和写入。 也就是说一个集合与同一变量上的任何后续get相关联。 原子compareAndSet方法也具有 这些内存一致性功能。 AtomicIntegerArray本质上是对int[]类型的封装。使用Unsafe类通过CAS的方式控制int[]在多线程下的安全 性。它提供了以下几个核心API //获得数组第i个下标的元素 public final int get(int i) //获得数组的长度 public final int length() //将数组第i个下标设置为newValue并返回旧的值 public final int getAndSet(int i, int newValue) //进行CAS操作如果第i个下标的元素等于expect则设置为update设置成功返回true public final boolean compareAndSet(int i, int expect, int update) //将第i个下标的元素加1 public final int getAndIncrement(int i) //将第i个下标的元素减1 public final int getAndDecrement(int i) //将第i个下标的元素增加deltadelta可以是负数 public final int getAndAdd(int i, int delta) 下面给出一个简单的示例展示AtomicIntegerArray使用 public class AtomicIntegerArrayDemo { static AtomicIntegerArray arr new AtomicIntegerArray(10); public static class AddThread implements Runnable{ public void run(){ for(int k0;k10000;k) arr.getAndIncrement(k%arr.length()); } } public static void main(String[] args) throws InterruptedException { Thread[] tsnew Thread[10]; for(int k0;k10;k){ ts[k]new Thread(new AddThread()); } for(int k0;k10;k){ts[k].start();} for(int k0;k10;k){ts[k].join();} System.out.println(arr); } } 输出结果 [10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000] 上述代码第2行申明了一个内含10个元素的数组。第3行定义的线程对数组内10个元素进行累加操作每个 元素各加1000次。第11行开启10个这样的线程。因此可以预测如果线程安全数组内10个元素的值必然都 是10000。反之如果线程不安全则部分或者全部数值会小于10000。 ➢ AtomicLong、AtomicLongArray原子性整型数组 AtomicLong、AtomicLongArray的API跟AtomicInteger、AtomicIntegerArray在使用方法都是差不多的。 区别在于用前者是使用原子方式更新的long值和long数组后者是使用原子方式更新的Integer值和Integer数组。 两者的相同处在于它们此类确实扩展了 Number允许那些处理基于数字类的工具和实用工具进行统一访问。在实际 开发中它们分别用于不同的场景。这个就具体情况具体分析了下面将举例说明 AtomicLong 的使用场景使用 AtomicLong生成自增长ID其他就不在过多介绍。 import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.concurrent.atomic.AtomicLong; public class AtomicLongTest { /** * param args */ public static void main(String[] args) { final AtomicLong orderIdGenerator new AtomicLong(0); final ListItem orders Collections .synchronizedList(new ArrayListItem()); for (int i 0; i 10; i) { Thread orderCreationThread new Thread(new Runnable() { public void run() { for (int i 0; i 10; i) { long orderId orderIdGenerator.incrementAndGet(); Item order new Item(Thread.currentThread().getName(), orderId); orders.add(order); } } }); orderCreationThread.setName(Order Creation Thread i); orderCreationThread.start(); } try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } SetLong orderIds new HashSetLong(); for (Item order : orders) { orderIds.add(order.getID()); System.out.println(Order name: order.getItemName() ----Order id: order.getID()); } } } class Item { String itemName; long id; Item(String n, long id) { this.itemName n; this.id id; } public String getItemName() { return itemName; } public long getID() { return id; }
}
输出
Order name:Order Creation Thread 0----Order id:1
Order name:Order Creation Thread 1----Order id:2
Order name:Order Creation Thread 0----Order id:4
Order name:Order Creation Thread 1----Order id:5
Order name:Order Creation Thread 3----Order id:3
Order name:Order Creation Thread 0----Order id:7
Order name:Order Creation Thread 1----Order id:6
........
Order name:Order Creation Thread 2----Order id:100
从运行结果我们看到不管是哪个线程。它们获得的 ID 是不会重复的保证的 ID 生成的原子性避免了线程安
全上的问题。
3) java.util.concurrent.lock包 待续...
