Java分布式应用学习笔记04JDK的并发包的集合总结.doc

Java分布式应用学习笔记04JDK的并发包的集合总结刘岩Email:1. 前言平时咱们使用的HashMap、ArrayList等等容器集合包都存在线程安全的问题，看过JDK源码的各位朋友们知道这些实现类底层，为了性能，都没有对这些集合的操作方法做加锁或者副本传递机制，只有Vector和Stack是线程安全的，大家可以看它们的源码，底层方法是以在方法上加上synchronized作为代价的，换句话说是用时间换取空间的方式。Sun JDK对多线程并发环境下做了很多并发的解决方案，其类大都在java.util.concurrent.*下面，此包下的类和java.util.*包下面的集合类，在使用上几乎没什么太大分别，想想也是啊！他们都是实现接口规范：List、Set、Map的。只要接口规范不变，那么在使用上也不应该有何变化，实现机制是一个侧重低概率并发或者就是单线程环境下，并发包则侧重高并发情况的系统。大家可以看看Tomcat的源代码，其中org.apache.catalina.core.ApplicationContext里面就使用到了并发包，因为Tomcat作为Web容器一定要接受来自各个客户端的request，进而分配Web应用上下文信息，应用参数key-value值等等。又得满足并发的请求、又得满足性能所需，所以它使用JDK的并发包。在使用层面上，笔者并不作过多介绍，可以参考非并发包的使用。至于这些非并发包的底层实现方式可以参考笔者的blog关于Java基础数据结构的基础知识，而是介绍一下这些并发包的底层机制和性能对比，在多线程环境下，用并发包和不用并发包的时间效率对比，空间资源效率不用比了，肯定单线程那些包消耗的比多线程消耗的小得多，毕竟做任何事都是要付出代价的。/blog/1004128。2. Map的并发包Map接口在并发包下的实现叫做java.util.concurrent.ConcurrentHashMap。它实现了ConcurrentMap接口，而ConcurrentMap接口又是继承自Map接口的扩展。先看看它是如何实现put操作的。 public V put(K key, V value) if (value = null) throw new NullPointerException(); int hash = hash(key.hashCode(); return segmentFor(hash).put(key, hash, value, false); 首先判断值是否为空，空值不必要存储，之后根据key的哈希值计算一个hash值。根据计算出的hash值去获取segment对象。找到了segment对象后调用该对象的put方法完成操作。Segment是ConcurrentHashMap的内部类其底层原理使用一个transient volatile HashEntry table;进行存取。现在再看segment内的put源码 V put(K key, int hash, V value, boolean onlyIfAbsent) lock(); try int c = count; if (c+ threshold) / ensure capacity rehash(); HashEntry tab = table; int index = hash & (tab.length - 1); HashEntry first = tabindex; HashEntry e = first; while (e != null & (e.hash != hash | !key.equals(e.key) e = e.next; V oldValue; if (e != null) oldValue = e.value; if (!onlyIfAbsent) e.value = value; else oldValue = null; +modCount; tabindex = new HashEntry(key, hash, first, value); count = c; / write-volatile return oldValue; finally unlock(); 首先是进行加锁操作，之后就是进行数组大小的判断，如果容量不够，则需要扩充。之后再通过对hash值的按位与的操作后，得到了这个key所要放置的位置。有了位置了，再看HashEntry数组组成的对象链，是否已经有key，如果有了，覆盖value操作，如果没有，创建一个新的HashEntry对象，重新组成HashEntry链表，最后进行解锁操作。所以直线我们关心的在put中会出现的线程安全问题，看了源码后是不是就解决了。想想除了put操作会出现线程不安全的隐患外，我们来看看remove操作。删除操作代码原理与put操作类似，也是通过hash值找到那个segment对象，之后调用segment的remove方法去完成真正的操作。真正的操作也是先加锁，之后迭代HashEntry，直到找到了传入的hash值相同的。找到了删之，重新建立链表！找不到，over，然后释放对象锁。在ConcurrentHashMap的get、containsKey等等这种不破坏原子性的读取（read）操作可以说大部分情况下没有进行加锁操作，即便像get加了锁操作，也是极其轻量的，仅仅是加锁了一行很简单的读取操作代码，如下 V readValueUnderLock(HashEntry e) lock(); try return e.value; finally unlock(); 下面我们来看看性能，在此所说的性能仅仅指时间执行效率。使用一般HashMap包的程序如下package threadConcurrent.hashMap;import java.util.HashMap;import java.util.Map;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/* author liuyan * */public class PubHashMap implements Runnable final static int ThreadSIZE = 2500;final static int elSize = 500;int threadNum;public PubHashMap(int threadNum) this.threadNum = threadNum;Overridepublic void run() Map hashMap = new HashMap();for (int i = 0; i elSize; i+) hashMap.put(i + + threadNum, i + + threadNum);/* * param args */public static void main(String args) / 启用线程池ExecutorService exec = Executors.newFixedThreadPool(ThreadSIZE);long startTime = System.currentTimeMillis();for (int index = 0; index = ThreadSIZE; index+) exec.execute(new PubHashMap(index);long endTime = System.currentTimeMillis();exec.shutdown();System.out.println(消耗时间： + (endTime - startTime) + ms);启动2500个线程，每个线程往HashMap中添加500个字符串元素。执行多次后给出一个平均时间吧消耗时间：1753ms使用并发包程序如下package threadConcurrent.hashMap;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/* author liuyan * */public class PutConcurrentHashMap implements Runnable final static int ThreadSIZE = 2500;final static int elSize = 500;int threadNum;public PutConcurrentHashMap(int threadNum) this.threadNum = threadNum;Overridepublic void run() Map concurrentHashMap = new ConcurrentHashMap();for (int i = 0; i elSize; i+) concurrentHashMap.put(i + + threadNum, i + + threadNum);/* * param args */public static void main(String args) / 启用线程池ExecutorService exec = Executors.newFixedThreadPool(ThreadSIZE);long startTime = System.currentTimeMillis();for (int index = 0; index = ThreadSIZE; index+) exec.execute(new PutConcurrentHashMap(index);long endTime = System.currentTimeMillis();exec.shutdown();System.out.println(消耗时间： + (endTime - startTime) + ms);也是多次执行后，得出一个平均时间吧消耗时间：1869ms时间消耗上差不多哈。在集合元素越来越多的情况下，在解决线程安全的同时保证了时间执行熬费上几乎和非线程安全的Map持平。所以在并发条件下不必自己解决Map的线程安全问题，直接放心使用JDK自己的并发Map包即可，时间性能上还能保证。3. List的并发包可以在高并发环境下使用java.util.concurrent.CopyOnWriteArrayList代替java.util.ArrayList。对于添加元素的操作，底层并不像Map那么复杂，就是利用了数组的copy功能和加锁机制public boolean add(E e) final ReentrantLock lock = this.lock;lock.lock();try Object elements = getArray();int len = elements.length;Object newElements = Arrays.copyOf(elements, len + 1);newElementslen = e;setArray(newElements);return true; finally lock.unlock();它是使用ReentrantLock进行的加锁，之后获得数组进行copy操作，之后数组个数加一。将新元素填充，之后再对局部变量进行一下set操作，最后就是解锁操作。至于remove操作，和add的原理一样public E remove(int index) final ReentrantLock lock = this.lock;lock.lock();try Object elements = getArray();int len = elements.length;Object oldValue = elementsindex;int numMoved = len - index - 1;if (numMoved = 0)setArray(Arrays.copyOf(elements, len - 1);else Object newElements = new Objectlen - 1;System.arraycopy(elements, 0, newElements, 0, index);System.arraycopy(elements, index + 1, newElements, index,numMoved);setArray(newElements);return (E) oldValue; finally lock.unlock();在枷锁对儿中间，先找到标记下的数组元素，之后创建一个新的临时数组，进行copy，将要删除的对象元素剔除出去！返回被删除元素对象。做add操作性能与ArrayList进行对比，线程运行400个，添加元素数是2000个。对比平均之后发现运行的时间也相差不是很多。并发情况下，CopyOnWriteArrayList比ArrayList略快了那么一点点。get几乎和ArrayList没差别，直接从数组中找第index个元素。4. Set的并发CopyOnWriteArraySet和CopyOnWriteArrayList底层实现差不多，就是在添加元素的时候需要对对象进行唯一性判断，如果对象数组已经含有重复的元素，不进行增加处理。在此不再赘述。5. Queue的并发队列的并发类是java.util.concurrent.ArrayBlockingQueue，从类名字上大家估计就能猜出来了，底层使用的依然是数组。这个ArrayBlockingQueue是继承自原始的java.util.AbstractQueue，所以很多方法在父类里面已经有了，只是对于关键方法入队列、出队列操作加入了锁对儿机制。入队列元素操作源码如下 public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException if (e = null) throw new NullPointerException();long nanos = unit.toNanos(timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try for (;) if (count != items.length) insert(e); return true; if (nanos = 0) return false; try nanos = notFull.awaitNanos(nanos); catch (InterruptedException ie) notFull.signal(); / propagate to non-interrupted thread throw ie; finally lock.unlock(); 数组未满情况下，执行insert操作的时候，如果数组满了，则进行等待，单位是纳秒。如果超时或者被唤醒了，那么再次判断是否数组已满，如果线程被打断直接抛出异常。出队列方法和入队列差不多，不再赘述。6. AtomicXXXX的原子类并发包还提供了支持原子操作的Atomic系列类，我们举一个具有代表性的类AtomicInteger类，通常我们使用计数器操作的时候，一般为了避免线程安全的问题，在方法上加锁操作。有了并发包下的原子系列类，我们直接使用即可。关键使用代码片段如下public static int getSum() return sum.incrementAndGet();其自增方法底层片段最关键是这么一句 if (compareAndSet(current, next) r