实战java高并发程序设计第三章(二)

3. JDK的并发容器

• 并发集合
  ConcurrentHashMap:这是一个高效的并发HashMap.你可以把它理解为一个线程安全的HashMap。
  CopyOnWriteArrayList:这是一个List，从名字看就知道它和ArrayList是一族的。在读多写少的场合，这个List的性能非常好，远远优于Vector。
  ConcurrentLinkedQueue:高效的并发队列，使用链表实现。可以看作一个线程安全的LinkedList.
  BlockingQueue:这是一个接口，JDK内部通过链表、数组等方式实现了这个接口。表示阻塞队列，非常适合作为数据共享的通道。
  ConcurrentSkipListMap:跳表的实现。这是一个Map，使用跳表的数据结构进行快速查找。
• 线程安全的HashMap
  可用Collections类来使普通HashMap转为线程安全的map

Collections.synchronizedMap(new HashMap())

private static class SynchronizedMap<K,V>
        implements Map<K,V>, Serializable {
        private static final long serialVersionUID = 1978198479659022715L;
        private final Map<K,V> m;     // 传入的map
        final Object      mutex;        // 锁资源对象,对map的任何操作都会锁该对象
        SynchronizedMap(Map<K,V> m) {
            this.m = Objects.requireNonNull(m);
            mutex = this;
        }
        SynchronizedMap(Map<K,V> m, Object mutex) {
            this.m = m;
            this.mutex = mutex;
        }
        public int size() {
            synchronized (mutex) {return m.size();}
        }
        public boolean isEmpty() {
            synchronized (mutex) {return m.isEmpty();}
        }
        public boolean containsKey(Object key) {
            synchronized (mutex) {return m.containsKey(key);}
        }
        public boolean containsValue(Object value) {
            synchronized (mutex) {return m.containsValue(value);}
        }
        public V get(Object key) {
            synchronized (mutex) {return m.get(key);}
        }

        public V put(K key, V value) {
            synchronized (mutex) {return m.put(key, value);}
        }
        public V remove(Object key) {
            synchronized (mutex) {return m.remove(key);}
        }
        public void putAll(Map<? extends K, ? extends V> map) {
            synchronized (mutex) {m.putAll(map);}
        }
        public void clear() {
            synchronized (mutex) {m.clear();}
        }
        .......  //省略
  }

• List的线程安全

Collections.synchronizedList(new LinkedList<String>())

• 高效读写队列ConcurrentLinkedQueue类

高并发环境中性能最好的队列,主要是利用CAS进行无锁操作,非阻塞队列

首先我们来看下它的Node节点:

private static class Node<E> {
        volatile E item;            //当前对象
        volatile Node<E> next;      //下一个对象,以此来构建链表
        Node(E item) {
            UNSAFE.putObject(this, itemOffset, item);
        }

        boolean casItem(E cmp, E val) {        //(期望值,设置目标值),cas操作
            return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
        }

        void lazySetNext(Node<E> val) {
            UNSAFE.putOrderedObject(this, nextOffset, val);
        }

        boolean casNext(Node<E> cmp, Node<E> val) {
            return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
        }
        private static final sun.misc.Unsafe UNSAFE;
        private static final long itemOffset;
        private static final long nextOffset;
        static {
            try {
                UNSAFE = sun.misc.Unsafe.getUnsafe();
                Class<?> k = Node.class;
                itemOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("item"));
                nextOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("next"));
            } catch (Exception e) {
                throw new Error(e);
            }
        }
    }

ConcurrentLinkedQueue类内部的tail指针更新并不是实时的,可能存在拖延现象,每次更新跳跃两个元素,如下图:

然后再看一下新增节点offer()方法:

public boolean offer(E e) {
        checkNotNull(e);        //非空校验
        final Node<E> newNode = new Node<E>(e);
        for (Node<E> t = tail, p = t;;) {    //for循环 无出口,知道设置成功
            Node<E> q = p.next;        //获取tail节点的next对象
            if (q == null) {        //第一次插入,p.next对象为空
                // p 为最后一个节点
                if (p.casNext(null, newNode)) {     //插入新元素,此时p=t
                    //每两次更新tail
                    if (p != t)         
                        casTail(t, newNode);  
                    return true;
                }
                // cas竞争失败,再次循环
            }
            else if (p == q)    //遇到哨兵
                // We have fallen off list.  If tail is unchanged, it
                // will also be off-list, in which case we need to
                // jump to head, from which all live nodes are always
                // reachable.  Else the new tail is a better bet.
                p = (t != (t = tail)) ? t : head;
            else
                // Check for tail updates after two hops.
                p = (p != t && t != (t = tail)) ? t : q; //t!=(t=tail) !=并不是原子操作,先取左边t的值,再取右边t=tail
        }
    }

• 高效读取:不变模式下的CopyOnWriteArrayList类

使用场景:读操作远远大于写操作,读操作越快越好,写操作慢一些也没事

特点:读取不用加锁,写入不会阻塞读取操作,只有写入和写入需要同步等待,读性能大幅提升

原理:写入时进行一次自我复制,修改内容写入副本中,写完后,再用副本内容替代原来的数据

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);  //进行复制
            newElements[len] = e;    //新数组代替老数组
            setArray(newElements);
            return true;
        } finally {
            lock.unlock();
        }
    }

4. JMH性能测试