并发编程之：数据结构

时间 2019-11-19

标签并发编程数据结构繁體版

原文原文链接

1.并发List

并发list有三种使用模式java

第一种使用Collections.synchronizedList方法进行包装：

List arrayList = new ArrayList();
//使用Collections.synchronizedList方法进行包装
List list = Collections.synchronizedList(arrayList);

咱们进入该包装方法内部查看源码发现，其实是有新建立了一个SynchronizedList对象，而SynchronizedList类对接对原有ArrayList类对象进行了包装，并对原对象的操做加上了同步控制。源码以下：数组

static class SynchronizedList<E>
    extends SynchronizedCollection<E>
    implements List<E> {
    private static final long serialVersionUID = -7754090372962971524L;

    final List<E> list;

    SynchronizedList(List<E> list) {
        super(list);
        this.list = list;
    }
    SynchronizedList(List<E> list, Object mutex) {
        super(list, mutex);
        this.list = list;
    }

    public boolean equals(Object o) {
        if (this == o)
            return true;
        synchronized (mutex) {return list.equals(o);}
    }
    public int hashCode() {
        synchronized (mutex) {return list.hashCode();}
    }

    public E get(int index) {
        synchronized (mutex) {return list.get(index);}
    }
    public E set(int index, E element) {
        synchronized (mutex) {return list.set(index, element);}
    }
    public void add(int index, E element) {
        synchronized (mutex) {list.add(index, element);}
    }
    public E remove(int index) {
        synchronized (mutex) {return list.remove(index);}
    }

    public int indexOf(Object o) {
        synchronized (mutex) {return list.indexOf(o);}
    }
    public int lastIndexOf(Object o) {
        synchronized (mutex) {return list.lastIndexOf(o);}
    }

    public boolean addAll(int index, Collection<? extends E> c) {
        synchronized (mutex) {return list.addAll(index, c);}
    }

    public ListIterator<E> listIterator() {
        return list.listIterator(); // Must be manually synched by user
    }

    public ListIterator<E> listIterator(int index) {
        return list.listIterator(index); // Must be manually synched by user
    }

    public List<E> subList(int fromIndex, int toIndex) {
        synchronized (mutex) {
            return new SynchronizedList<>(list.subList(fromIndex, toIndex),
                                        mutex);
        }
    }

    @Override
    public void replaceAll(UnaryOperator<E> operator) {
        synchronized (mutex) {list.replaceAll(operator);}
    }
    @Override
    public void sort(Comparator<? super E> c) {
        synchronized (mutex) {list.sort(c);}
    }

    /**
     * SynchronizedRandomAccessList instances are serialized as
     * SynchronizedList instances to allow them to be deserialized
     * in pre-1.4 JREs (which do not have SynchronizedRandomAccessList).
     * This method inverts the transformation.  As a beneficial
     * side-effect, it also grafts the RandomAccess marker onto
     * SynchronizedList instances that were serialized in pre-1.4 JREs.
     *
     * Note: Unfortunately, SynchronizedRandomAccessList instances
     * serialized in 1.4.1 and deserialized in 1.4 will become
     * SynchronizedList instances, as this method was missing in 1.4.
     */
    private Object readResolve() {
        return (list instanceof RandomAccess
                ? new SynchronizedRandomAccessList<>(list)
                : this);
    }
}

第二种使用List的同步实现类Vector：

咱们经过查看Vector源码发现，Vector类实现了List接口，并实现的接口方法都是同步的方式实现,因为Vector中的实现方法是同步方法必然会对元素存储效率产生严重影响。如get/set方法代码片断安全

/**
 * Returns the element at the specified position in this Vector.
 *
 * @param index index of the element to return
 * @return object at the specified index
 * @throws ArrayIndexOutOfBoundsException if the index is out of range
 *            ({@code index < 0 || index >= size()})
 * @since 1.2
 */
public synchronized E get(int index) {
    if (index >= elementCount)
        throw new ArrayIndexOutOfBoundsException(index);

    return elementData(index);
}

/**
 * Replaces the element at the specified position in this Vector with the
 * specified element.
 *
 * @param index index of the element to replace
 * @param element element to be stored at the specified position
 * @return the element previously at the specified position
 * @throws ArrayIndexOutOfBoundsException if the index is out of range
 *         ({@code index < 0 || index >= size()})
 * @since 1.2
 */
public synchronized E set(int index, E element) {
    if (index >= elementCount)
        throw new ArrayIndexOutOfBoundsException(index);

    E oldValue = elementData(index);
    elementData[index] = element;
    return oldValue;
}

/**
 * Appends the specified element to the end of this Vector.
 *
 * @param e element to be appended to this Vector
 * @return {@code true} (as specified by {@link Collection#add})
 * @since 1.2
 */
public synchronized boolean add(E e) {
    modCount++;
    ensureCapacityHelper(elementCount + 1);
    elementData[elementCount++] = e;
    return true;
}

第三种使用List实现类CopyOnWriteArrayList：

CopyOnWriteArrayList的内部实现与Vector不一样，当对象进行读操做时，直接返回结果，操做中不进行同步控制；当进行写操做时会在该方法上面加上锁，而后复制一个副本列表，对副本进行修改，最后将副本写回原来列表。数据结构

其get/set/add方法源码片断以下多线程

/**
 * {@inheritDoc}
 *
 * @throws IndexOutOfBoundsException {@inheritDoc}
 */
public E get(int index) {
    return get(getArray(), index);
}

/**
 * Replaces the element at the specified position in this list with the
 * specified element.
 *
 * @throws IndexOutOfBoundsException {@inheritDoc}
 */
public E set(int index, E element) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] elements = getArray();
        E oldValue = get(elements, index);

        if (oldValue != element) {
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len);
            newElements[index] = element;
            setArray(newElements);
        } else {
            // Not quite a no-op; ensures volatile write semantics
            setArray(elements);
        }
        return oldValue;
    } finally {
        lock.unlock();
    }
}

/**
 * Appends the specified element to the end of this list.
 *
 * @param e element to be appended to this list
 * @return {@code true} (as specified by {@link Collection#add})
 */
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();
    }
}

老是所知CopyOnWriteArrayList只是对修改进行了同步控制，因此在读操做上面比上面两种方式要快一些。并发

2.并发set

第一种方式也是有Collections.synchronizedSet方法进行包装：

Set hashSet = new HashSet();
//Collections.synchronizedSet进行包装
Set set = Collections.synchronizedSet(hashSet);

包装以后新生产一个包装了原对象的SynchronizedSet对象，SynchronizedSet对象的读写操做仍是对原有的set进行读写，只是在包装的读写方法上面加上了同步控制。app

第二种和list比较类似，并发set也有一个CopyOnWriteArraySet实现：

它实现了set接口，而且是线程安全的。它的内部实现彻底依赖于CopyOnWriteArrayList咱们能够查看源码dom

public class CopyOnWriteArraySet<E> extends AbstractSet<E>
        implements java.io.Serializable {
    private static final long serialVersionUID = 5457747651344034263L;

    private final CopyOnWriteArrayList<E> al;

    /**
     * Creates an empty set.
     */
    public CopyOnWriteArraySet() {
        al = new CopyOnWriteArrayList<E>();
    }

    /**
     * Creates a set containing all of the elements of the specified
     * collection.
     *
     * @param c the collection of elements to initially contain
     * @throws NullPointerException if the specified collection is null
     */
    public CopyOnWriteArraySet(Collection<? extends E> c) {
        if (c.getClass() == CopyOnWriteArraySet.class) {
            @SuppressWarnings("unchecked") CopyOnWriteArraySet<E> cc =
                (CopyOnWriteArraySet<E>)c;
            al = new CopyOnWriteArrayList<E>(cc.al);
        }
        else {
            al = new CopyOnWriteArrayList<E>();
            al.addAllAbsent(c);
        }
    }

    /**
     * Returns the number of elements in this set.
     *
     * @return the number of elements in this set
     */
    public int size() {
        return al.size();
    }

    /**
     * Returns {@code true} if this set contains no elements.
     *
     * @return {@code true} if this set contains no elements
     */
    public boolean isEmpty() {
        return al.isEmpty();
    }

    public boolean contains(Object o) {
        return al.contains(o);
    }

    /**
     */
    public Object[] toArray() {
        return al.toArray();
    }

    /**
     */
    public <T> T[] toArray(T[] a) {
        return al.toArray(a);
    }

    /**
     */
    public void clear() {
        al.clear();
    }

    /**
     */
    public boolean remove(Object o) {
        return al.remove(o);
    }

    /**
     */
    public boolean add(E e) {
        return al.addIfAbsent(e);
    }

    /**
     */
    public boolean containsAll(Collection<?> c) {
        return al.containsAll(c);
    }

    /**
     */
    public boolean addAll(Collection<? extends E> c) {
        return al.addAllAbsent(c) > 0;
    }

    /**
     */
    public boolean removeAll(Collection<?> c) {
        return al.removeAll(c);
    }

    /**
     */
    public boolean retainAll(Collection<?> c) {
        return al.retainAll(c);
    }

    /**
     */
    public Iterator<E> iterator() {
        return al.iterator();
    }

    /**
     */
    public boolean equals(Object o) {
        if (o == this)
            return true;
        if (!(o instanceof Set))
            return false;
        Set<?> set = (Set<?>)(o);
        Iterator<?> it = set.iterator();

        // Uses O(n^2) algorithm that is only appropriate
        // for small sets, which CopyOnWriteArraySets should be.

        //  Use a single snapshot of underlying array
        Object[] elements = al.getArray();
        int len = elements.length;
        // Mark matched elements to avoid re-checking
        boolean[] matched = new boolean[len];
        int k = 0;
        outer: while (it.hasNext()) {
            if (++k > len)
                return false;
            Object x = it.next();
            for (int i = 0; i < len; ++i) {
                if (!matched[i] && eq(x, elements[i])) {
                    matched[i] = true;
                    continue outer;
                }
            }
            return false;
        }
        return k == len;
    }

    public boolean removeIf(Predicate<? super E> filter) {
        return al.removeIf(filter);
    }

    public void forEach(Consumer<? super E> action) {
        al.forEach(action);
    }

    /**
     */
    public Spliterator<E> spliterator() {
        return Spliterators.spliterator
            (al.getArray(), Spliterator.IMMUTABLE | Spliterator.DISTINCT);
    }

    /**
     */
    private static boolean eq(Object o1, Object o2) {
        return (o1 == null) ? o2 == null : o1.equals(o2);
    }
}

3.并发Map

一样适用Map时也能够经过Collections.synchronizedMap进行包装实现一个线程安全的Map，可是这不是最优的实现方式，因为Map是一个咱们适用至关频繁的数据结构，因此JDK为提供了一个专用于高并发的Map实现ConcurrentHashMap，它的内部实现了锁分离，为多线程并发下高性能提供了保证。ide

ConcurrentHashMap是专门为线程并发设计的Map，它的get方法是无锁的，它的put方法操做的锁力度又小于同步的map因此总体性能高于同步的。高并发

4.并发Queue

在并发队列上，JDK提供了两套实现，一个是以ConcurrentListQueue为表明的高性能队列，一个是以BlockingQueue为表明的阻塞队列。他们都继承自Queue接口。

ConcurrentListQueue：

是一个适合于高并发场景下的队列。它经过无锁的方式，实现了高并发状态下的高性能。

BlockingQueue：

是阻塞队列BlockingQueue的表明实现，主要应用场景是生产者消费者的共享数据。因为LinkedBlockingQueue提供一种阻塞等待的机制，好比当消费者速度快于生产者时队列中的消息很快就会被清空，此时消费者再从队列中取数据时就会被阻塞等待。同理当生产者速度快于消费者时，队列很快就会被放满数据此时再往队列里面放数据时生产者就会被阻塞等待。

BlockingQueue队列主要实现方法：

offer(object):将object添加到BlockingQueue里，若是队列有足够空间就返回true，不然返回false（该方法不会阻塞当前执行的线程）。

offer(object,timeout,timeunit):是阻塞队列特有的方法，将object添加到BlockingQueue里，若是队列有足够空间则当即返回true，若是队列没有足够空间则堵塞timeout时间段，若是还添加不进去返回fase。

put(object):把object添加到队里，若是队列已满则该线程会阻塞直到队列有空间加进去。

poll():从队列里面取出元素，若是没有返回null，该方法不会阻塞线程。

poll(timeout,timeunit):从队列里面取出元素。若是队里里面没有元素会阻塞timeout时间段，若是超时尚未数据可取则返回null。

take():从队列里面取出元素，若是队列里面没有元素则线程会阻塞直到有新元素加入队列。

drainTo(list):将队列里面的数据一次性所有取出到list里面，该方法能够一次性取出全部数据到list中不用进行堵塞，能够提升数据提取效率。

ArrayBlockingQueue：

是一种基于数组的阻塞队列实现，内部维护了一个定长数组用于换成数据对象。另外还保存着两个整形变量分别标记着队列的头部和尾部在数组中的位置。

LinkedBlockingQueue：

是一种基于链表的阻塞队列实现。

5.并发Deque（Double-Ended Queue）双端队列

deque是一个能够在队列的头部或尾部进行入队和出队操做，新增操做以下：

有三个实现类：LinkedList、ArrayDeque、LinkedBlockingDeque。

LinkedList使用链表实现的双端队列，ArrayDeque使用数组实现的双端队列。一般状况下因为ArrayDeque基于数组实现，具备高效的随机访问性能，所以ArrayDeque具备更好的遍历性能。可是当队列大小变化较大时ArrayDeque须要从新分配内存，进行数组复制，在这种状况下基于链表的LinkedList具备更好的表现。可是不管LinkedList或是ArrayDeque都不是线程安全的。

LinkedBlockingDeque是一个线程安全的双端队列实现。LinkedBlockingDeque是一个链表结构，每个队列节点都维护一个前驱节点和一个后驱节点。LinkedBlockingDeque没有进行读写锁分离，所以在高并发使用中其性能要低于LinkedBlockingQueue，更低于ConCurrentLinkedQueue。