多线程六同步容器&并发容器

亦凉 2021-10-23 17:58 356阅读 0赞

### 同步容器(使用的是synchronized,并且不一定是百分百安全) ###

> **本篇续 -- 线程之间的通信 ,介绍java提供的并发集合,既然正确的使用wait和notify比较困难,java平台为我们提供了更高级的并发容器来替代**

### 一. Vector&ArrayList ###

*  Vector虽然它的set和get方法都被Synchronized修饰,但是开启两条线程并发访问,一条线程拼命往里写,另一台循环往移除,这样并发访问不一定是百分百的线程安全的,很可能出现数组越界异常,而且现在基本已经不适用它了,它基本被ArrayList替代掉了
 *  ArrayList是线程不安全的

将ArrayList转换成线程安全的

ArrayList al = new ArrayList(); Connections.syncchronizedList(al);

### Map ###

*  **效率低下的HashTable,同Vector类似,它被HashMap替代掉了**
    
    > HashTable底层是通过synchronized关键字来实现的,虽然线程安全,但是在高并发的情况下,效率却超级低,因为当一个线程访问HashTable的同步方法时,其他线程拿不到锁,就访问不了它的同步方法,比如线程1使用HashTable的put方法,其他线程不仅不能使用put,就连get()也被阻塞**不可以存储null键值**
 *  **线程不安全的HashMap**
    
    > 在高并发访问的情况下,不会使用HashMap,**线程不安全,但是可以存储null键值**

将HashMap转换成线程安全的

HashMap map = new HashMap<>(); Map map1 = Collections.synchronizedMap(map);

--------------------

# 并发容器**J.U.C** #

### 一. 并发List--CopyOnWriteArrayList ###

> 它是ArrayList 的一个线程安全的变体，其中所有可变操作（add、set 等等）都是通过对底层数组进行一次新的复制来实现的。

*  我们着重的关注点是啥呢? 可能会出现线程安全性问题的方法 **set() add() remove()**,

### 基本的使用; ###

> 同样实现了List接口,那么其实它的使用和ArrayList完全一样的,只是内部的实现不一样

### 实现原理 ###

**分析add方法**

/** * 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; //获取锁ReentrantLock lock.lock(); try { Object[] elements = getArray(); // 获取CopyOnWriteArrayList内部维护的数组 int len = elements.length; Object[] newElements = Arrays.copyOf(elements, len + 1); // 复制该数组到 newElements数组里面 并扩容 newElements[len] = e; //在数组最后添加新的元素 setArray(newElements); //让替换掉原来的数组 return true; //添加成功 } finally { lock.unlock(); } } /** * {@inheritDoc} * * @throws IndexOutOfBoundsException {@inheritDoc} */ public E get(int index) { return get(getArray(), index); }

> 可以看到,其实类似读写分离, add()可能会出现线程安全问题,因此给它一把锁,get()不会出现线程安全性问题,因此没有锁,如果非要给get也加上锁,那么在add的时候,就不能get了,因为它拿不到锁对象

**之所以说是读写分离,读read,读取的原数组,add写的时候,其实不是往原array里面写,而是分如下几步**

1.  加锁
2.  拷贝原数组,创建新的数组
3.  添加新的元素
4.  新数组替换原数组
5.  释放锁

**分析remove方法**

/** * Removes the element at the specified position in this list. * Shifts any subsequent elements to the left (subtracts one from their * indices). Returns the element that was removed from the list. * * @throws IndexOutOfBoundsException {@inheritDoc} */ public E remove(int index) { final ReentrantLock lock = this.lock; lock.lock(); try { Object[] elements = getArray(); int len = elements.length; E oldValue = get(elements, index); int numMoved = len - index - 1; if (numMoved == 0) //移除最后一个 setArray(Arrays.copyOf(elements, len - 1)); else { // 移除其它的,当前元素后面的需要往前移动 Object[] newElements = new Object[len - 1]; System.arraycopy(elements, 0, newElements, 0, index); System.arraycopy(elements, index + 1, newElements, index, numMoved); setArray(newElements); } return oldValue; // 移除谁,返回谁 } finally { lock.unlock(); } }

**总结一下**

*  假如说多线程并发的应用中,**绝大部分都是读操作**,那么,**CopyOnWriteArrayList效率明显高**,因为他是读写分离,读没有锁,是在原数组上进行的,写也是线程安全的(加了lock),其中有一点就是它虽然能保证数据的最终是同步的,但是却保证不了实时同步性
 *  假如说**绝大部分操作是写操作**,可以看到,还是挺吃内存的,数组过大,他的效率可能就不一定比同步容器高,在不知道往里面存储多少数据的情况下,慎用.在高并发的互联网环境下这种操作分分钟就导致故障

--------------------

### 二. 并发Set ###

*  和List步调一致的是,java平台为**set集合提供了CopyOnWriteArraySet**,它实现了set接口,**底层完全依赖CopyOnWriteArrayList**因此,它的特性和CopyOnWriteArrayList一致,同样在**读多写少**的高并发环境下,拥有很高的效率  
    
 *  同样,在**写多读少**的高并发环境下,我们可以考虑下面的转换

Set<Object> set = Collections.synchronizedSet(new HashSet<>());

### 三. 并发Map ###

*  同样可以使用Collections获取到一个同步Map,但是这个Map的性能依然不是最优的

Collections.synchronizedMap();

*  jdk同样提供了一个高效同步Map,**ConcurrentHashMap** 他不允许空值
    
    > **jdk1.7中ConcurrentHashMap内部实行了锁分离**,分段式存储数据,然后每一段数据都会加上不同步的锁,所以当其中一条线程访问其中一段数据的时候,其他数据仍然可以被别的线程访问,同时,它的get()方法也是无锁的

> **jdk1.8中ConcurrentHashMap内部抛弃了锁分离而使用红黑树实现**

### 四. 并发Queue ###

> 在并发队列上,java提供了两套实现,一个是ConcurrentLinkedQueue的非阻塞型的队列,另一个是BlockingQueue接口,阻塞队列,同样他们继承了Queue接口

### 1 . ConcurrentLinkQueue 非阻塞队列 ###

> 保证了在高并发的情况下,对link底层维护的链表的增删改各个节点的安全性

*  API和LinkList等完全一样...

#### 实现原理 ####

> **它通过无锁的方法,底层使用的是UNSAFE实现(保证线程的安全性)**  
> **入队offer**

/** * Inserts the specified element at the tail of this queue. * As the queue is unbounded, this method will never return {@code false}. * * @return {@code true} (as specified by {@link Queue#offer}) * @throws NullPointerException if the specified element is null */ 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; if (q == null) { // p is last node if (p.casNext(null, newNode)) { // Successful CAS is the linearization point // for e to become an element of this queue, // and for newNode to become "live". if (p != t) // hop two nodes at a time casTail(t, newNode); // Failure is OK. //可以看到,在这种会出现线程安全性问题的地方,使用的是cas进行操作,保证了线程的安全性 return true; } // Lost CAS race to another thread; re-read next } 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; } } // 可以看到,casXXX 底层使用的是UNSAFE实现(保证线程的安全性) private boolean casTail(Node<E> cmp, Node<E> val) { return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val); } private boolean casHead(Node<E> cmp, Node<E> val) { return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val); }

# 图解节点的添加过程,在001文件中 #

**出队**

public E poll() { restartFromHead: for (;;) { for (Node<E> h = head, p = h, q;;) { E item = p.item; if (item != null && p.casItem(item, null)) { // Successful CAS is the linearization point // for item to be removed from this queue. if (p != h) // hop two nodes at a time updateHead(h, ((q = p.next) != null) ? q : p); return item; } else if ((q = p.next) == null) { updateHead(h, p); return null; } else if (p == q) continue restartFromHead; else p = q; } }

--------------------

### 阻塞队列BlockingQueue ###

java.util.concurrent  
接口 BlockingQueue  
类型参数：  
E - 在此 collection 中保持的元素类型  
所有超级接口：  
Collection, Iterable, Queue  
所有已知子接口：  
BlockingDeque  
所有已知实现类：  
ArrayBlockingQueue, DelayQueue, LinkedBlockingDeque, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue

主要应用场景:  
**生产者和消费者模式**

**BlockingQueue是一个接口,因此我们学习ArrayBlockingQueue**,它的底层维护着一个数组的阻塞队列

> 一个由数组支持的有界阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序。队列的头部 是在队列中存在时间最长的元素。队列的尾部 是在队列中存在时间最短的元素。新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素。

> 此类支持对等待的生产者线程和使用者线程进行排序的可选公平策略。默认情况下，不保证是这种排序。然而，通过将公平性 (fairness) 设置为 true 而构造的队列允许按照 FIFO 顺序访问线程。公平性通常会降低吞吐量，但也减少了可变性和避免了“不平衡性”。

> 此类及其迭代器实现了 Collection 和 Iterator 接口的所有可选 方法。

既然叫阻塞队列,也就是说,他支持在多线程的条件下,**多线程并发添加移除数组的元素,会被阻塞等待,而不会抛出空值异常或者报错,且数组的长度不可变**

*  两种情况发生阻塞
    
     *  队列满的时候,进行入队操作,也就是说,队列满了,但是有一个线程往队列里面put的时候,他会被阻塞,除非有别的线程做了take的操作
     *  队列为空,进行出队操作

<table> 
 <thead> 
  <tr> 
   <th>可阻塞方法</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>put</td> 
   <td>将指定元素插入此队列中，<strong>将等待可用的空间（如果有必要）。</strong></td> 
  </tr> 
  <tr> 
   <td>take</td> 
   <td>获取并移除此队列的头部，<strong>在元素变得可用之前一直等待（如果有必要）。</strong></td> 
  </tr> 
 </tbody> 
</table>

> 可以直接使用它,实现消费者生产者模式,他的底层是用Condition ReentrantLock实现的--,方法被lock()和unlock()锁住, 数组满就await , 出队后signal

<table> 
 <thead> 
  <tr> 
   <th>不发生阻塞</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>add</td> 
   <td>将指定元素插入此队列中（如果立即可行且不会违反容量限制），成功时返回 true，如果当前没有可用的空间，则抛出 <strong>IllegalStateException</strong></td> 
  </tr> 
  <tr> 
   <td>remove</td> 
   <td>从此队列中移除指定元素的单个实例（如果存在）。同样会抛异常</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <thead> 
  <tr> 
   <th>不发生阻塞</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>offer</td> 
   <td>将指定元素插入此队列中（如果立即可行且不会违反容量限制），成功时返回 true，<strong>如果当前没有可用的空间，则返回 false。</strong></td> 
  </tr> 
  <tr> 
   <td>poll</td> 
   <td>获取并移除此队列的头部，在指定的等待时间前等待可用的元素（如果有必要）。<strong>没有元素,返回null</strong></td> 
  </tr> 
 </tbody> 
</table>

--------------------

### 五 .并发Deque ###

jdk1.6开始java提供的双端队列**Deque(Double Ended Queue)**允许在队列的头部或者尾部进行入队或者出队的操作.  
他是是实现类:

*  ArrayDeque
    
     *  使用数组实现了双端队列,拥有更好的随机访问性,但是当队列的增大时,他需要重新分配内存,然后进行数组的复制
 *  LinkedList
    
     *  使用链表实现了双端队列,因此它相对于ArrayDeque来说,没有了内存调整,数组复制的负担
        
        > **无论是ListedList还是ArratDeque,都是线程不安全的**
 *  LinkedBlockingDeque
    
    ## > 线程安全,但是它没有进行读写分离,也就说,同一时间,只允许一条线程对其操作,因此在并发中,**它的性能,远远底于ConcurrentLinkQueue** ##
    
    **参考 博文**  
    [你不就像风一样][Link 1]

转载于:https://www.cnblogs.com/ZhuChangwu/p/11150328.html

[Link 1]: https://www.cnblogs.com/yueshutong/p/9696216.html