Java并发编程-锁及并发容器

时间 2019-11-29

标签 java 并发编程容器栏目 Java 繁體版

原文原文链接

锁

锁是用来控制多个线程访问共享资源的方式，通常来讲，一个锁可以防止多个线程同时访问共享资源（可是有些锁能够容许多个线程并发的访问共享资源，好比读写锁）。在Lock接口出现以前，Java程序是靠synchronized关键字实现锁功能的，而JavaSE5以后，并发包中新增了Lock接口（以及相关实现类）用来实现锁功能，它提供了与synchronized关键字相似的同步功能，只是在使用时须要显式地获取和释放锁。虽然它缺乏了（经过synchronized块或者方法所提供的）隐式获取释放锁的便捷性，可是却拥有了锁获取与释放的可操做性、可中断的获取锁以及超时获取锁等多种synchronized关键字所不具有的同步特性。html

重入锁ReentrantLock

重入锁ReentrantLock，顾名思义，就是支持重进入的锁，它表示该锁可以支持一个线程对资源的重复加锁。除此以外，该锁的还支持获取锁时的公平和非公平性选择。ReentrantLock是java.unti.concurrent包下的一个类，它的通常使用结构以下所示：java

public void lockMethod() {  
    ReentrantLock myLock = new ReentrantLock();  
    myLock.lock();  
    try{  
        // 受保护的代码段  
        //critical section  
    } finally {  
        // 能够保证发生异常 锁能够获得释放 避免死锁的发生  
        myLock.unlock();  
    }  
}复制代码

ReentrantLock与synchronized的比较

相同：ReentrantLock提供了synchronized相似的功能和内存语义。
不一样：

ReentrantLock功能性方面更全面，好比时间锁等候，可中断锁等候，锁投票等，所以更有扩展性。在多个条件变量和高度竞争锁的地方，用ReentrantLock更合适，ReentrantLock还提供了Condition，对线程的等待和唤醒等操做更加灵活，一个ReentrantLock能够有多个Condition实例，因此更有扩展性。
ReentrantLock 的性能比synchronized会好点。
ReentrantLock提供了可轮询的锁请求，他能够尝试的去取得锁，若是取得成功则继续处理，取得不成功，能够等下次运行的时候处理，因此不容易产生死锁，而synchronized则一旦进入锁请求要么成功，要么一直阻塞，因此更容易产生死锁。

公平性

在Java的ReentrantLock构造函数中提供了两种锁：建立公平锁和非公平锁（默认）。代码以下：node

public ReentrantLock() {

       sync = new NonfairSync();

}

 public ReentrantLock(boolean fair) {
          sync = fair ? new FairSync() : new NonfairSync();
    }复制代码

在公平的锁上，线程按照他们发出请求的顺序获取锁，但在非公平锁上，则容许‘插队’：当一个线程请求非公平锁时，若是在发出请求的同时该锁变成可用状态，那么这个线程会跳过队列中全部的等待线程而得到锁。算法

非公平锁性能高于公平锁性能的缘由：
在恢复一个被挂起的线程与该线程真正运行之间存在着严重的延迟。编程

读写锁ReentrantReadWriteLock

以前提到锁基本都是排他锁，这些锁在同一时刻只容许一个线程进行访问，而读写锁在同一时刻能够容许多个读线程访问，可是在写线程访问时，全部的读线程和其余写线程均被阻塞。读写锁维护了一对锁，一个读锁和一个写锁，经过分离读锁和写锁，使得并发性相比通常的排他锁有了很大提高。数组

通常状况下，读写锁的性能都会比排它锁好，由于大多数场景读是多于写的。在读多于写
的状况下，读写锁可以提供比排它锁更好的并发性和吞吐量。Java并发包提供读写锁的实现是ReentrantReadWriteLock缓存

public class Cache {
static Map<String, Object> map = new HashMap<String, Object>();
static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
static Lock r = rwl.readLock();
static Lock w = rwl.writeLock();
// 获取一个key对应的value
public static final Object get(String key) {
r.lock();
try {
return map.get(key);
} finally {
r.unlock();
}
}
// 设置key对应的value，并返回旧的value
public static final Object put(String key, Object value) {
w.lock();
try {
return map.put(key, value);
} finally {
w.unlock();
}
}
// 清空全部的内容
public static final void clear() {
w.lock();
try {
map.clear();
} finally {
w.unlock();
}
}
}复制代码

Cache组合一个非线程安全的HashMap做为缓存的实现，同时使用读写锁的
读锁和写锁来保证Cache是线程安全的。在读操做get(String key)方法中，须要获取读锁，这使
得并发访问该方法时不会被阻塞。写操做put(String key,Object value)方法和clear()方法，在更新
HashMap时必须提早获取写锁，当获取写锁后，其余线程对于读锁和写锁的获取均被阻塞，而
只有写锁被释放以后，其余读写操做才能继续。安全

Condition接口

Condition是在java 1.5中才出现的，它用来替代传统的Object的wait()、notify()实现线程间的协做，相比使用Object的wait()、notify()，使用Condition的await()、signal()这种方式实现线程间协做更加安全和高效。bash

调用Condition的await()和signal()方法，都必须在lock保护以内，就是说必须在lock.lock()和lock.unlock之间才可使用多线程

Conditon中的await()对应Object的wait()
Condition中的signal()对应Object的notify()
Condition中的signalAll()对应Object的notifyAll()

public class ConTest {  

     final Lock lock = new ReentrantLock();  
     final Condition condition = lock.newCondition();  

    public static void main(String[] args) {  
        // TODO Auto-generated method stub  
        ConTest test = new ConTest();  
        Producer producer = test.new Producer();  
        Consumer consumer = test.new Consumer();  


        consumer.start();   
        producer.start();  
    }  

     class Consumer extends Thread{  

            @Override  
            public void run() {  
                consume();  
            }  

            private void consume() {  

                    try {  
                           lock.lock();  
                        System.out.println("我在等一个新信号"+this.currentThread().getName());  
                        condition.await();  

                    } catch (InterruptedException e) {  
                        // TODO Auto-generated catch block  
                        e.printStackTrace();  
                    } finally{  
                        System.out.println("拿到一个信号"+this.currentThread().getName());  
                        lock.unlock();  
                    }  

            }  
        }  

     class Producer extends Thread{  

            @Override  
            public void run() {  
                produce();  
            }  

            private void produce() {                   
                    try {  
                           lock.lock();  
                           System.out.println("我拿到锁"+this.currentThread().getName());  
                            condition.signalAll();                             
                        System.out.println("我发出了一个信号："+this.currentThread().getName());  
                    } finally{  
                        lock.unlock();  
                    }  
                }  
     }  

}复制代码

执行结果：

我在等一个新信号Thread-1
我拿到锁Thread-0
我发出了一个信号：Thread-0
拿到一个信号Thread-1复制代码

并发容器

CopyOnWrite容器

CopyOnWrite容器即写时复制的容器。通俗的理解是当咱们往一个容器添加元素的时候，不直接往当前容器添加，而是先将当前容器进行Copy，复制出一个新的容器，而后新的容器里添加元素，添加完元素以后，再将原容器的引用指向新的容器。这样作的好处是咱们能够对CopyOnWrite容器进行并发的读，而不须要加锁，由于当前容器不会添加任何元素。因此CopyOnWrite容器也是一种读写分离的思想，读和写不一样的容器

在使用CopyOnWriteArrayList以前，咱们先阅读其源码了解下它是如何实现的。如下代码是向ArrayList里添加元素，能够发如今添加的时候是须要加锁的，不然多线程写的时候会Copy出N个副本出来。

public boolean add(T 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();

    }

}

final void setArray(Object[] a) {
    array = a;
}复制代码

读的时候不须要加锁，若是读的时候有多个线程正在向ArrayList添加数据，读仍是会读到旧的数据，由于写的时候不会锁住旧的ArrayList。

public E get(int index) {
    return get(getArray(), index);
}复制代码

ConcurrentHashMap的实现原理与使用

ConcurrentHashMap是线程安全且高效的HashMap。ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成。Segment是一种可重入锁ReentrantLock，在ConcurrentHashMap里扮演锁的角色，HashEntry则用于存储键值对数据。一个ConcurrentHashMap里包含一个Segment数组，Segment的结构和HashMap相似，是一种数组和链表结构，一个Segment里包含一个HashEntry数组，每一个HashEntry是一个链表结构的元素，每一个Segment守护者一个HashEntry数组里的元素,当对HashEntry数组的数据进行修改时，必须首先得到它对应的Segment锁。

为何要使用ConcurrentHashMap

线程不安全的HashMap

在多线程环境下，使用HashMap进行put操做会引发死循环，致使CPU利用率接近100%，因此在并发状况下不能使用HashMap

效率低下的HashTable

HashTable容器使用synchronized来保证线程安全，但在线程竞争激烈的状况下HashTable的效率很是低下。由于当一个线程访问HashTable的同步方法，其余线程也访问HashTable的同步方法时，会进入阻塞或轮询状态。如线程1使用put进行元素添加，线程2不但不能使用put方法添加元素，也不能使用get方法来获取元素，因此竞争越激烈效率越低

ConcurrentHashMap的锁分段技术可有效提高并发访问率

HashTable容器在竞争激烈的并发环境下表现出效率低下的缘由是全部访问HashTable的线程都必须竞争同一把锁，假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不一样数据段的数据时，线程间就不会存在锁竞争，从而能够有效提升并发访问效率，这就是ConcurrentHashMap所使用的锁分段技术。首先将数据分红一段一段地存储，而后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其余段的数据也能被其余线程访问

hash定位

在定位元素的代码里咱们能够发现，定位HashEntry和定位Segment的散列算法虽然同样，都与数组的长度减去1再相“与”，可是相“与”的值不同，定位Segment使用的是元素的hashcode经过再散列后获得的值的高位，而定位HashEntry直接使用的是再散列后的值。其目的是避免两次散列后的值同样，虽然元素在Segment里散列开了，可是却没有在HashEntry里散列开

hash >>> segmentShift) & segmentMask　　// 定位Segment所使用的hash算法
int index = hash & (tab.length - 1);　　// 定位HashEntry所使用的hash算法复制代码

get

Segment的get操做实现很是简单和高效。先通过一次再散列，而后使用这个散列值经过散
列运算定位到Segment，再经过散列算法定位到元素，代码以下

public V get(Object key) {
int hash = hash(key.hashCode());
return segmentFor(hash).get(key, hash);
}复制代码

get操做的高效之处在于整个get过程不须要加锁，除非读到的值是空才会加锁重读。咱们
知道HashTable容器的get方法是须要加锁的，那么ConcurrentHashMap的get操做是如何作到不加锁的呢？缘由是它的get方法里将要使用的共享变量都定义成volatile类型，如用于统计当前Segement大小的count字段和用于存储值的HashEntry的value。定义成volatile的变量，可以在线程之间保持可见性，可以被多线程同时读，而且保证不会读到过时的值

put

因为put方法里须要对共享变量进行写入操做，因此为了线程安全，在操做共享变量时必须加锁。put方法首先定位到Segment，而后在Segment里进行插入操做。插入操做须要经历两个步骤，第一步判断是否须要对Segment里的HashEntry数组进行扩容，第二步定位添加元素的位置，而后将其放在HashEntry数组里

public V put(K key, V value) {
        Segment<K,V> s;
        if (value == null)
            throw new NullPointerException();
        int hash = hash(key);
        int j = (hash >>> segmentShift) & segmentMask;
        if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
             (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
            s = ensureSegment(j);
        return s.put(key, hash, value, false);
    }复制代码

Segment的put方法

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
            HashEntry<K,V> node = tryLock() ? null :
                scanAndLockForPut(key, hash, value);
            V oldValue;
            try {
                HashEntry<K,V>[] tab = table;
                int index = (tab.length - 1) & hash;
                HashEntry<K,V> first = entryAt(tab, index);
                for (HashEntry<K,V> e = first;;) {
                    if (e != null) {
                        K k;
                        if ((k = e.key) == key ||
                            (e.hash == hash && key.equals(k))) {
                            oldValue = e.value;
                            if (!onlyIfAbsent) {
                                e.value = value;
                                ++modCount;
                            }
                            break;
                        }
                        e = e.next;
                    }
                    else {
                        if (node != null)
                            node.setNext(first);
                        else
                            node = new HashEntry<K,V>(hash, key, value, first);
                        int c = count + 1;
                        if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                            rehash(node);
                        else
                            setEntryAt(tab, index, node);
                        ++modCount;
                        count = c;
                        oldValue = null;
                        break;
                    }
                }
            } finally {
                unlock();
            }
            return oldValue;
        }复制代码

size

ConcurrentHashMap的作法是先尝试2次经过不锁住Segment的方式来统计各个Segment大小，若是统计的过程当中，容器的count发生了变化，则再采用加锁的方式来统计全部Segment的大小。
那么ConcurrentHashMap是如何判断在统计的时候容器是否发生了变化呢？使用modCount变量，在put、remove和clean方法里操做元素前都会将变量modCount进行加1，那么在统计size先后比较modCount是否发生变化，从而得知容器的大小是否发生变化

阻塞队列

阻塞队列（BlockingQueue）是一个支持两个附加操做的队列。这两个附加的操做支持阻塞的插入和移除方法。

支持阻塞的插入方法：意思是当队列满时，队列会阻塞插入元素的线程，直到队列不
满。
支持阻塞的移除方法：意思是在队列为空时，获取元素的线程会等待队列变为非空。阻塞队列经常使用于生产者和消费者的场景，生产者是向队列里添加元素的线程，消费者是从队列里取元素的线程。阻塞队列就是生产者用来存放元素、消费者用来获取元素的容器

插入和移除操做的4中处理方式

方法/处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除方法	remove()	poll()	take()	poll(e,time,unit)
检查方法	element()	peek()	不可用	不可用

抛出异常：当队列满时，若是再往队列里插入元素，会抛出IllegalStateException（"Queue
full"）异常。当队列空时，从队列里获取元素会抛出NoSuchElementException异常。
返回特殊值：当往队列插入元素时，会返回元素是否插入成功，成功返回true。若是是移
除方法，则是从队列里取出一个元素，若是没有则返回null。
一直阻塞：当阻塞队列满时，若是生产者线程往队列里put元素，队列会一直阻塞生产者
线程，直到队列可用或者响应中断退出。当队列空时，若是消费者线程从队列里take元素，队
列会阻塞住消费者线程，直到队列不为空。
超时退出：当阻塞队列满时，若是生产者线程往队列里插入元素，队列会阻塞生产者线程
一段时间，若是超过了指定的时间，生产者线程就会退出

Java里的阻塞队列

JDK 7提供了7个阻塞队列，以下。

ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列。

ArrayBlockingQueue是一个用数组实现的有界阻塞队列。此队列按照先进先出（FIFO）的原则对元素进行排序。默认状况下不保证线程公平的访问队列

LinkedBlockingQueue：一个由链表结构组成的有界阻塞队列。

LinkedBlockingQueue是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。

PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列。

PriorityBlockingQueue是一个支持优先级的无界阻塞队列。默认状况下元素采起天然顺序升序排列。也能够自定义类实现compareTo()方法来指定元素排序规则，或者初始化PriorityBlockingQueue时，指定构造参数Comparator来对元素进行排序。须要注意的是不能保证同优先级元素的顺序

DelayQueue：一个使用优先级队列实现的无界阻塞队列。

DelayQueue是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队列中的元素必须实现Delayed接口，在建立元素时能够指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素

SynchronousQueue：一个不存储元素的阻塞队列。

SynchronousQueue是一个不存储元素的阻塞队列。每个put操做必须等待一个take操做，不然不能继续添加元素

LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。

LinkedTransferQueue是一个由链表结构组成的无界阻塞TransferQueue队列。相对于其余阻塞队列，LinkedTransferQueue多了tryTransfer和transfer方法。

LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。

LinkedBlockingDeque是一个由链表结构组成的双向阻塞队列。所谓双向队列指的是能够从队列的两端插入和移出元素。双向队列由于多了一个操做队列的入口，在多线程同时入队时，也就减小了一半的竞争。相比其余的阻塞队列，LinkedBlockingDeque多了addFirst、addLast、offerFirst、offerLast、peekFirst和peekLast等方法

参考资料

《Java并发编程的艺术》

ReentrantLock之公平锁与非公平锁浅析

java condition使用及分析

聊聊并发-Java中的Copy-On-Write容器

Java集合---ConcurrentHashMap原理分析