java并发编程分析总结

深碍√TFBOYSˉ_ 2023-07-22 09:09 13阅读 0赞

# 总的来说，java并发编程要掌握的大概为一下几个方面： #

## java内存模型 ##

## java中的锁 ##

## java并发容器 ##

## java并发工具类 ##

## java Executor框架 ##

本人最近也在学习中，所以会不时更新上面五个方面的学习总结，废话不多说，这就开干。

1.java内存模型

这一块需要掌握几个点：

java内存特性

volatile关键字

happens-before规则

既然讲到内存模型肯定得需要一张图吧。看图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvbGZsaWtlaW5ub2NlbmNl_size_16_color_FFFFFF_t_70][]

这张图其实就是想说明，每个线程读取数据并不是直接和主内存打交道，而是有自己的工作内存，有了自己的工作内存，那么主内存的共享变量的一致性、可见性问题就非常重要了。

由此引出内存模型三大特性：

内存一致性：主内存共享变量对所有线程必须一致

内存可见性：指的是某个线程对主内存的共享变量必须让所有线程知道

内存原子性：对于一个共享变量操作必须是原子操作

volatile关键字：

volatile特性：

可见性

原子性

volatile写-读建立的happens-before关系：  
A线程在写volatile变量之前所有可见的共享变量，在B线程读同一个volatile变量后，将立即变得对B线程可见

volatile写内存语义：

当写一个volatile变量时，JMM会把该线程对应的本地内存中的共享变量值刷新到主内存

volatile读的内存语义：

当读一个volatile变量时，JMM会把线程对应的本地内存置为无效。线程接下来将从主内存中读取共享变量

volatile读写内存语义总结：

线程A写一个volatile变量，实质上是线程A向接下来将要读这个volatile变量的某个线程发出了（其对共享变量所做修改的）消息

线程B读一个volatile变量，实质上是线程B接收了之前某个线程发出的（在写这个volatile变量之前对共享变量所做修改的）消息

线程A写一个volatile变量，随后线程B读这个volatile变量，这个过程实质上是线程A通过主内存向线程B发送消息

总结：volatile实现了禁止指令重排序和保证变量可见性

volatile内存语义的实现：

为了实现volatile的内存语义，编译器在生成字节码时，会在指令序列中插入内存屏障来禁止特定类型的处理器重排序。

锁的释放和获取的内存语义：

当线程释放锁时，JMM会把线程对应的本地内存中的共享变量刷新到主内存中

当线程获取锁时，JMM会把线程对应的本地内存中的共享变量置为无效

锁释放和锁获取总结：

线程A释放一个锁，实质上是线程A向接下来将要获取这个锁的某个线程发出了（线程A对共享变量所做修改的）消息

线程B获取一个锁，实质上是线程B接收了之前某个线程发出的（在释放这个锁之前对共享变量所做修改的）消息

线程A释放锁，随后线程B获取这个锁，这个过程实质上是线程A通过主内存向线程B发送消息

锁内存语义的实现：

volatile内存语义是靠添加内存屏障实现的，而锁内存语义本质上是借助volatile实现的，可想而知volatile非常重要。比如可重入锁ReentrantLock，在调用ReentrantLock.lock()的时候，其实就是改变一个volatile变量的操作，导致其他线程调用lock()函数的时候发生阻塞。由于这是在多线程环境中，因此同步多个线程的原始变量就不能含糊，必须做到一致性、可见性，这个原始变量就是同步器AQS维护的volatile变量state。非公平锁释放与公平锁释放一样，最后都是要写volatile变量，但是非公平锁获取的方式不是读volatile，而是通过CAS方式（相信大家对CAS比较熟悉，我就不介绍了），CAS简单来说是三步骤：读、比较、写，这三步骤是一起完成的，本质上来讲也可以说是把这三步骤加锁了，可是他并没有用java的锁对象加锁，而是通过硬件条件加锁的，所以由低层锁实现的CAS成为了许多java锁对象的底层操作。总的来说，锁底层实现就是靠CAS和volatile（volatile也是通过底层硬件手段实现的）。

对比锁释放的内存语义与volatile写-读的内存语义：

锁释放与volatile写有相同的内存语义；锁获取与volatile读有相同的内存语义。

happens-before规则：如果一个操作的执行的结果需要对另一个操作可见，那么这两个操作之间必须要存在happens-before规则。

这个规则其实我们刚才应用到volatile的内存语义和锁的内存语义上了，就是volatile写在volatile读之前，锁释放在锁获取之前（这两句看起来可能有点懵，解释一下，这里的volatile写-读和锁释放-获取不是一个线程所为，而是一个线程进行volatile写或者锁释放，随后另一个线程进行volatile读或者锁获取，这其实体现的正是线程之间的同步）

## java锁 ##

我们都知道java线程之间的同步无非就是用synchronized关键字和java锁实现，但是synchronized有时候达不到我们想要的结果，简而言之就是synchronized功能不强大，粒度太大，不够灵活，因此java锁就应运而生了。

锁按不同的维度有不同的分法：

按一个线程的多个流程能不能获取同一把锁：

可重入锁：ReentrantLock，其实就是同步器AQS中的状态state可以取大于1的数，利用计数法来实现线程再次进入锁

不可重入锁：Mutex，同步器AQS中的状态state只能取0或1

按多个线程争取锁是是否需要排队：

公平锁：由于同步器AQS是有阻塞队列的，先来的线程争取锁失败后先进队，等上个线程释放锁时，自然先进队的线程先获取锁

非公平锁：利用CAS操作不断循环操作state变量，谁抢到了，谁就获取锁了

（ReetrantLock有公平锁和非公平锁两种实现）

按多个线程可不可以共享同一把锁：

共享锁：说明state状态可以同时由多个线程修改

独占锁：state状态只能0或1，且只有一个线程在共享区域

还有好多种类型，我就不分了，例如悲观锁、乐观锁啥的，这其实说的还是java锁和CAS，我们就记住java的所有锁都是由volatile和CAS实现的，而volatile和CAS是由底层硬件实现的。

下面来说一说锁实现另外一个重要的数据结构：AQS（AbstractQueueSynchronizer），队列同步器

这个数据结构非常重要，因为他是完成所功能的核心，他是锁的一个内部类，该类维护一个volatile共享变量state（我们之前一直提的），该类完成了对这个state的各种操作以及维护同步队列，继承该同步器在实现的时候只需要实现共享资源state的获取和释放方式。

我们先来分析AQS自己实现的几个比较重要的方法：

ReetrantLock：

protected final boolean tryAcquire(int acquires) {
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) {
                    if (!hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) {
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
                else if (current == getExclusiveOwnerThread()) {
                    int nextc = c + acquires;
                    if (nextc < 0)
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                return false;
            }

这个方法是内部类自定义同步器的获取锁方法，也可以说是volatile读,注意else if中的nextc = c + acquires,这个意思就是当前线程如果还要获取锁那就把状态再加1（acquires=1），这就是计数法。

再看看释放锁：

protected final boolean tryRelease(int releases) {
                int c = getState() - releases;
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                if (c == 0) {
                    free = true;
                    setExclusiveOwnerThread(null);
                }
                setState(c);
                return free;
            }

注意最后setState(c)这条语句，这就是在释放锁的时候volatile写，当然前面依然是使用计数法c = getState() - releases

## java并发容器 ##

讲到并发容器我就讲讲ConcurrentHashMap和ConcurrentLinkedQueue，其实只要分析出一个并发容器的源码，其他容器分析起来很容易的。

讲ConcurrentHashMap肯定要先讲讲HashMap，首先我们都知道java1.7的HashMap和1.8的HashMap是不一样的，1.7的数据结构就是hash数组加链表实现的，而1.8就是在原来基础上把链表改为链表+红黑树组合，这是为了链表过长导致访问速度慢，然后就规定当链表长到阈值后改为访问时间复杂度为o(logn)的红黑树。HashMap这个数据结构是非常优秀的，访问和修改的时间都为非常快，可是在多线程中会出现线程安全问题，比如会出现经典的环形链表，导致出现死循环。

看过好多关于死循环的文章，都讲得太啰嗦了。我把关键代码贴出来，一看便知晓：

public void transfer(Entry[] newTable){
            Entry[] src = table;
            int newCapacity = newTable.length;
            for(int j = 0;j<src.length;j++){
                Entry<K,V> e = src[j];
                if(e!=null){
                    src[j] = null;
                    do{
                        Entry<K,V> temp = e.next;
                        int i = indexFor(e.hashcode,newCapacity);
                        e.next = newTable[i];
                        newTable[i] = e;
                        e = temp;
                    }while (e!=null);
                }
            }
        }

问题就出现在这个方法上，我介绍一下流程：

1.首先执行put方法插入entry

2.执行addEntry进行插入操作

3.执行插入操作之前判断是否需要扩容，执行resize()

4.而扩容的本质就是申请比原来大两倍的空间（一般来说），然后把原来的元素全部复制过来，执行transfer()

就在执行transfer方法时可能会出现环形链表情况，

首先在同一条链表中，有A，B两个元素（插入时都是头插的，扩容复制的时候也是头插的）比如现在是A-->B-->NULL

线程1、线程2两个同时想扩容，所以都会拿到A，然后比如线程1先会进行头插，最后新表上会是：B-->A-->NULL,然后线程2再来进行头插，此时他拿到的是原来的A，他还是会拿A的引用去新表头插，可是此时新表的头是B，而不是NULL，所以最终新表上会是：A-->B-->A(为了方便才这样写的，这里A是同一个A)，这就导致出现了环形链表。

因此，这里无非是加个synchronized就完事了（HashTable）。

但是加个synchronized太重了，有没有轻一点的锁呢，有，那就是ConcurrentHashMap，1.7版本的思想是分段锁，提高并行度，也即是说可以存在同一时刻多个线程访问，主要思想是把共享区域换分为更小的共享区域，这样多个线程虽然不能同时访问同一个区域，但是可以访问不同的共享区域啊。1.8的思想是CAS操作，这个我暂时没有看，以后分析。

先来看看1.7版本的分段锁，核心类是Segment，继承自ReetrantLock，大概看看ConcurrentHashMap的结构：

public class ConcurrentHashMap<K,V>{
        
        static final class Segment<K,V> extends ReentrantLock implements Serializable {
    
        private static final long serialVersionUID = 2249069246763182397L;
    
            static final int MAX_SCAN_RETRIES =
                Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
    
            transient volatile HashEntry<K,V>[] table;
    
            transient int count;
    
            transient int modCount;
    
            transient int threshold;
    
            final float loadFactor;
        }
        static final class HashEntry<K,V> {
            final int hash;
            final K key;
            volatile V value;
            volatile HashEntry<K,V> next;
        }
        final Segment<K,V>[] segments;
    
    
    
    }

很明显，它有两个内部类，然后维护一个数组，每个数组就是一个段，每个段其实就是类似于Hashmap,只不过每个段都是线程安全的。

下面我们来分析多线程的put操作是如何实现的，

public V put(K key,V value){
            Segment<K,V> s;
            if(value == null) throw new NullPointerException();
            int hash = HashMap.hash(key);
            s = ensureSegment(hash);//通过hash值获取segment
            return s.put(key,hash,value,false);
        }

从这个put操作可以看出，其实第一步就是想办法通过hash值获取Segment(s = ensureSegment(hash))

然后在这个Segment里进行真正的put操作（前面说过其实Segment和hashmap很像），我们再来看看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; //同hashMap相同的哈希定位方式
                    HashEntry<K,V> first = entryAt(tab, index);
                    for (HashEntry<K,V> e = first;;) {
                        if (e != null) {
                           //若不为null，则持续查找，知道找到key和hash值相同的节点，将其value更新
                            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 { //若头结点为null
                            if (node != null) //在遍历key对应节点链时没有找到相应的节点
                                node.setNext(first);
                                //当前修改并不需要让其他线程知道，在锁退出时修改自然会
                                //更新到内存中,可提升性能
                            else
                                node = new HashEntry<K,V>(hash, key, value, first);
                            int c = count + 1;
                            if (c > threshold && tab.length < HashMap.max_capacity)
                                rehash(node); //如果超过阈值，则进行rehash操作
                            else
                                setEntryAt(tab, index, node);
                            ++modCount;
                            count = c;
                            oldValue = null;
                            break;
                        }
                    }
                } finally {
                    unlock();
                }
                return oldValue;
            }

既然到了Segment的put操作了，说明现在开始需要同步了，因为Segment是最小的共享区域了，需要要加锁了，所以put方法一开始就要获取锁了（tryLock()和scanAndLockPut()）,即使tryLock失败了，scanAndLockPut()方法有while循环tryLock(),直到获取到锁为止，接下来tryCatch部分就是对HashEntry操作了，最后释放锁。（关于tryCatch操作部分和hashmap是差不多的，一般都是先用key获取hash，hash获取桶位置，然后迭代这个桶，如果碰到相同key的就覆盖，不同的继续走下去，直到走到NULL，然后进行头插，头插时判断容量是否超了阈值，那就要进行扩容，扩容完后，把原先的引用指向新扩容的空间。）

现在再来说说ConcurrentLinkedQueue，我们都队列有列表和链表两种方式，队列（除了优先队列这种的）一般就是先进先出，所以列表对队列没有多大优势，链表可以有无界的优势。

ConcurrentLinkedQueue是一个基于链接节点的无界线程安全队列，使用CAS操作来实现，首先来看看他的入队：

public boolean offer(E e){
            if(e == null)throw new NullPointerException();
            Node<E> newNode = new Node<E>(e);
            for (Node<E> t = tail, p = t;;) {
                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.
                        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;
            }
        }

从代码可以看到，有两个重要的方法casNext()和casTail(),这两个方法可以说是该队列实现的核心。其次最难理解的是条件判断，因此可以和出队方法一起理解：

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;
                }
            }
        }

在队列里面应该有一下几种情况：

1.  先考虑单线的offer
2.  再考虑多线程时候的offer：
    
    1.  多个线程offer
    2.  部分线程offer，部分线程poll
        
        1.  poll比offer快
        2.  offer比poll快

关于以上几点情况很多，我也是看了好多文章才看懂，这里放一篇非常棒的博客，相信大家也会看懂的：[https://blog.csdn.net/u011521203/article/details/80214968][https_blog.csdn.net_u011521203_article_details_80214968]。

暂时到这里吧，这篇博客写的很粗糙，我会经常修改并且添加更多细节进去的。谢谢大家阅读

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvbGZsaWtlaW5ub2NlbmNl_size_16_color_FFFFFF_t_70]: /images/20230528/50ffe0ef47064f1f8eb3d77b855d321f.png
[https_blog.csdn.net_u011521203_article_details_80214968]: https://blog.csdn.net/u011521203/article/details/80214968