Java 多线程：同步集合与同步锁

同步集合
同步集合在多线程开发中扮演非常重要的角色，本文介绍些常用但被忽略的同步集合。

CopyOnWriteArrayList
Copy-On-Write 是一种用于程序设计中的优化策略，基本思路是多个线程共享同一个列表，当某个线程想要修改这个列表的元素时会把列表中的元素 Copy 一份，然后进行修改，修改完后再讲新的元素设置给这个列表，是一种延时懒惰策略。好处是可以对 CopyOnWrite 容器进行并发的读，而不需要加锁，因为当前容器不会添加、移除任何元素。所以 CopyOnWrite 容器也是一种读写分离的思想，读和写不同的容器。使用 Copy-On-Write 机制实现的并发容器有两个分别是：CopyOnWriteArrayList 和 CopyOnWriteArraySet。

下面来分析下 CopyOnWriteArrayList 的核心源码，首先看下 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();
    }
}

可以看到在添加的时候进行了加锁操作，否则多线程写的时候会 Copy 出 N 个副本出来。复制一份之后将新的元素设置到元素数组的 len 位置，然后再把最新的元素设置给该列表。

get 方法：

public E get(int index) {
    return get(getArray(), index);
}

读不需要加锁，如果读的时候多个线程正在向容器内添加数据，还是会读到旧数据，因为写的时候不会锁住旧的元素数组。

这种写时拷贝的原理优点是读写分离，并发场景下操作效率会提高，缺点是写操作时占用的内存空间翻了一倍，因此是以空间换时间。

ConcurrentHashMap
HashTable 是 HashMap 的线程安全实现，但是 HashTable 使用 synchronized 来保证线程安全，这就会导致它的效率非常低下，因为当线程 1 使用 put 添加元素，线程 2 不但不能使用 put 添加元素，同时也不能使用 get 获取元素，竞争越激烈效率越低。

因此替代 HashTable 的 ConcurrentHashMap 就出现了，ConcurrentHashMap 的优点在于容器里有多把锁，每一把锁用于锁容器其中一部分数据，当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效的提高并发访问效率。它的原理是将数据分成一段一段的存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据时，其他段的数据也能被其他线程访问。有些方法需要跨段，如 size()和 containsValue()，他们可能需要锁定整个表而不仅是某个段，这需要按顺序锁定所有段，操作完毕后又按顺序释放所有段的锁。

BlockingQueue
阻塞队列是生产者-消费者的一个实现，当队列满了时，再次调用 put 函数添加元素，那么调用线程将会阻塞，直到队列不再是填满状态。避免了手动判断以及同步操作。

函数名 作用
add(e) 把元素 e 添加到 BlockingQueue 里,如果 BlockingQueue 可以容纳,则返回 true,否则抛异常
offer(e) 把元素 e 添加到 BlockingQueue 里,如果 BlockingQueue 可以容纳,则返回 true,否则返回 false
offer(e,time,unit) 把元素 e 添加到 BlockingQueue 里,如果 BlockingQueue 可以容纳,则返回 true,否则在等待指定的时间之后继续尝试添加,如果失败则返回 false
put(e) 把元素 e 添加到 BlockingQueue 里,如果 BlockingQueue 不能容纳,则调用此方法的线程被阻塞直到 BlockingQueue 里面有空间再继续添加
take() 取走 BlockingQueue 里排在队首的对象,若 BlockingQueue 为空,则进入等待状态直到 BlockingQueue 有新的对象被加入为止
poll(time,unit) 取出并移除队列中的队首元素,如果设定的阻塞时间内还没有获得数据,那么返回 null
element() 获取队首元素,如果队列为空,那么抛出 NoSuchElementException 异常
peek() 获取队首元素,如果队列为空,那么返回 null
remove() 获取并移除队首元素,如果队列为空,那么抛出 NoSuchElementException 异常
BlockingQueue 多种常用实现:

ArrayBlockingQueue 数组实现的、线程安全的、有界的阻塞队列
按 FIFO（先进先出）原则对元素进行排序，元素从尾部插入到队列，从头部开始返回。
LinkedBlockingQueue 单向链表实现的队列
按 FIFO（先进先出）原则对元素进行排序，元素从尾部插入到队列，从头部开始返回。吞吐量高于 ArrayBlockingQueue，但是在大多数并发应用程序中其可预知的性能要低,功能类似的有 ConcurrentLinkedQueue
LinkedBlockingDeque 双向链表实现的双向并发阻塞队列
同时支持 FIFO 和 FILO，即可以从队列的头和尾同时操作（插入/删除），支持线程安全。可以指定队列容量（默认容量大小等于 Integer.MAX_VALUE）
同步锁
线程安全就是必须通过各种锁机制来进行同步，防止某个对象或者值在多个线程中被修改导致的不一致问题。为了保证数据的一致性，需要通过同步机制保证在同一时刻只有一个线程能够访问到该对象或者数据，修改完毕之后再将最新数据同步到主存中，使其他线程能够得到最新数据。

synchronized
Java 中最常用的同步机制就是 synchronized 关键字，它是一种基于语言的粗略锁，能够作用于对象、函数、类。每个对象都只有一个锁，谁拿到锁就得到了访问权限。

public class SynchronizedDemo {

// 只对SynchronizedDemo当前对象生效
public synchronized void syncMethod() {
}

public void syncThis() {
    // 只对SynchronizedDemo当前对象生效
    synchronized (this) {
    }
}

// 对SynchronizedDemo所有对象生效
public void syncClassMethod() {
    synchronized (SynchronizedDemo.class) {
    }
}

// 对SynchronizedDemo所有对象生效
public synchronized static void syncStaticMethod() {
}

}

上面例子分别演示了同步方法、同步块、同步 class 对象、同步静态方法。前两种锁的是对象，作用是防止其他线程同时访问同一个对象中的 synchronized 代码块或者函数。后两种锁的是 class 对象，作用是防止其他线程同时访问所有对象中的 synchronized 锁的代码块，因为 Class 锁对类的所有对象实例起作用。

ReentrantLock 与 Condition
Java5 之前协调共享对象访问时，只有 synchronized 和 volatile，Java6 增加了 ReentrantLock，与 synchronized 相比，实现了相同的语义，但具有更高的灵活性，并可以提供轮训锁和定时锁，同时可以提供公平锁或非公平锁。

函数 作用
lock() 获取锁
tryLock() 尝试获取锁
tryLock(long timeout,TimeUnit unit) 尝试获取锁，如果到了指定的时间还获取不到，那么超时
unLock() 释放锁
newCondition() 获取锁的 Condition
lock、tryLock 与 unlock 一般成对出现，用法如下：

Lock lock = new ReentrantLock();
public int doSth() {
    lock.lock();
    try {
        // do some thing
    } finally {
        lock.unlock();
    }
}

需要注意的是必须在 finally 块中释放 lock，否则如果代码抛出异常就永远释放不了锁。而使用 synchronized 锁，JVM 将确保锁会自动释放，并且当 JVM 使用 synchronized 管理锁定请求和释放时，JVM 在生成线程转储时能够包括锁定信息，这些对调试非常有价值，因为它们能标识死锁或者其他异常行为的来源。而 Lock 类只是普通的类，JVM 不知道具体哪个线程拥有 Lock 对象，这也是 Lock 没有完全替代掉 synchronized 的原因。

ReentrantLock 中还有一个重要函数 newCondition()，用于获取 Lock 上的 Condition，Condition 是用于实现线程间的通讯，解决 Object.wait()、notify()、nofityAll()难以使用的问题。

常用方法如下：

函数 作用
await() 线程等待
await(int time，TimeUnit unit) 线程等待特定时间，超过时间则为超时
signal() 随机唤醒某个等待线程
signalAll() 唤醒所有等待中的线程
下面通过 ReentrantLock 与 Condition 实现一个简单的阻塞队列，实现代码如下：

public class MyArrayBlockingQueue {
// 数据数组
private final T[] items;
// 锁
private final Lock lock = new ReentrantLock();
// 队满的条件
private Condition notFull = lock.newCondition();
// 队空条件
private Condition notEmpty = lock.newCondition();
// 头部索引
private int head;
// 尾部索引
private int tail;
// 数据的个数
private int count;

public MyArrayBlockingQueue(int maxSize) {
    items = (T[]) new Object[maxSize];
}

public MyArrayBlockingQueue() {
    this(10);
}

public void put(T t) {
    lock.lock();
    try {
        while (count == getCapacity()) {
            System.out.println("数据已满，等待");
            notFull.await();
        }
        items[tail] = t;
        if (++tail == getCapacity()) {
            tail = 0;
        }
        ++count;
        notEmpty.signalAll(); // 唤醒等待数据的线程
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        lock.unlock();
    }
}

public T take() {
    lock.lock();
    try {
        while (count == 0) {
            System.out.println("还没有数据，请等待");
            notEmpty.await();
        }
        T ret = items[head];
        items[head] = null;
        if (++head == getCapacity()) {
            head = 0;
        }
        --count;
        notFull.signalAll(); // 唤醒添加数据的线程
        return ret;
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        lock.unlock();
    }
    return null;
}

public int getCapacity() {
    return items.length;
}

public int size() {
    lock.lock();
    try {
        return count;
    } finally {
        lock.unlock();
    }
}

public static void main(String[] args) {
    MyArrayBlockingQueue<Integer> aQueue = new MyArrayBlockingQueue<Integer>();
    aQueue.put(3);
    aQueue.put(24);
    for (int i = 0; i < 5; i++) {
        System.out.println(aQueue.take());
    }
}

}

上面代码模拟了一个有界队列阻塞队列，阻塞条件分别使用 notfull 与 notEmpty，当调用 put 函数时集合元素已满那么会调用 notFull.await()堵塞调用线程，直到其他线程调用了 take()方法，由于 take 会在队列中取出一个元素后调用 notFull.signalAll()唤醒等待线程，使得 put 可以继续。同理 take 函数是当元素数量为 0 时调用 notEmpty.await()进行等待，当其他线程调用 put 方法执行 notEmpty.signalAll()才唤醒 take 函数的线程，使之能够取得元素

作者：提辖鲁
来源：CSDN
原文：https://blog.csdn.net/lj402159806/article/details/83051518?utm_source=copy