java 并发编程艺术摘抄

并发编程的挑战

并发执行不一定比串行执行快，因为线程有创建和上下文切换的开销。

减少上下文切换的方法有无锁并发编程、CAS算法、使用最少线程和使用协程。

• 无锁并发编程。多线程竞争锁会引起上下文切换，所以多线程处理数据时，可以使用一些办法来避免使用锁，如将数据的ID按照Hash算法取模分段，不同的线程处理不同的段。我也不懂怎么实现
• CAS 算法。java的 Atomic 包使用 CAS 算法来更新数据，而不需要加锁。
• 使用最少线程。避免不需要的线程，比如任务很少，但创建了很多线程来处理，这样会造成大量的线程都处于等待状态。
• 协程：在单线程里实现多任务的调度，并在单线程里维持多个任务之间的切换。

死锁

避免死锁的几个常见方法

• 避免一个线程同时获取多个锁。
• 避免一个线程在锁内同时占用多个资源 ，尽量保证每一个锁只占用一个资源
• 尝试使用定时锁，使用 lock.tryLock(timeout)来替代使用内部锁的机制
• 对于数据库锁，加锁和解锁必须在一个数据库里，否则会出现解锁失败的情况

资源限制

程序的执行速度受限于计算机硬件资源或者软件资源。如服务器的带宽只有 2mb/s,某个资源下载速度是1mb/s ，系统启动10个线程下载，下载速度不会加快。

我们需要在不同的环境下调整程序的并发度，如根据数据库连接数调整线程执行sql的数量。

java 并发机制的底层实现原理

Synchonized

• 对于普通同步方法锁的是当前实例对象
• 对于静态同步方法，锁的是当前类的Class对象
• 对于同步方法块，锁的是 Synchonized 括号里配置的对象。

java内存模型

在深入理解java 虚拟机中就有所涉猎，所以这里部分省略。

Valatile

这里让我意外的是作者在书中表明Valatile 具有原子性，原文：

原子性：对任意单个的 volatile 变量的读/写具有原子性，但类似于 volatile++ 这种复合操作不具有原子性。

java 并发基础

suspend()、resume()、和stop()方法不推荐使用

因为suspend()方法调用后线程不会释放已经占用的资源，而是占有资源进入睡眠状态，这样容易引发死锁。stop方法在终结一个线程时不会保证线程的资源正常释放。

安全地终止线程

我们可以使用线程1.interruput 的方法然后让线程自已判断是否需要再运行下去。

interrupt与isInterrupt

还可以使用 boolean 变量来控制是否需要停止任务并终止该线程。

public static void main(String[] args) throws InterruptedException {
        Thread countThread = new Thread(new Runner(), "countThread");
        countThread.start();
        TimeUnit.SECONDS.sleep(1);
        countThread.interrupt();
        Runner two = new Runner();
        countThread = new Thread(two, "CountThread");
        countThread.start();
        TimeUnit.SECONDS.sleep(1);
        two.cancel();
}
    private static class Runner implements Runnable{
        private long i;
        private volatile boolean on = true;
        @Override
        public void run() {
            while (on&&!Thread.currentThread().isInterrupted()){
                i++;
            }
            System.out.println("count i="+i);
        }
        public void cancel(){
            on = false;
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

输出

count i=401674570
count i=324268636
1
2

等待通知机制

一个线程修改了一个对象的值，而另一个线程感知到了变化，然后进行相应的操作，整个过程开始处于一个线程，而最终执行的又是一个线程。java的内置等待/通知 api 能够很好的解决这个问题

一个线程A调用了对象o的 wait() 方法进入等待状态，而另一个线程B 调用了对象 o 的 notify() 或者 notifyAll() 方法，线程A 收到通知后从对象 O 的wait() 方法返回，进而执行后续操作。

1. 使用 wait()、notify()、和 notifyAll()时需要先对调用对象加锁
2. 调用 wait() 方法后，线程由 Running 变成 Waiting，并将当前线程放置到对象的等待对队列
3. notify() 或者 notifyAll() 方法调用后，等待线程依旧不会从 wait() 返回，需要调用 notify() 或者 notifyAll() 的线程释放锁之后，等待线程才有机会从 wait() 中返回。
4. notify() 方法将等待队列中的一个等待线程从等待队列中移到同步队列中,而notifyAll()方法是将等待队列中所有的线程全部移到同步队列,被移动的线程状态从 Waiting 变为 Blocked
5. 从 wait() 方法返回的前提是获得了调用对象的锁

等待通知经典范式，即加锁、条件循环和处理逻辑3个步骤。

Thread.join()的使用

如果一个线程执行了 thread.join() 语句其含义是当前线程A 等待 thread 线程终止后才从 thread.join() 返回

/**
 * 现在有T1、T2、T3三个线程，你怎样保证T2在T1执行完后执行，T3在T2执行完后执行？
 */
public class JoinDemo {

    public static void main(String[] args) {
        Thread t1=new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("t1 is running...");
            }
        });

        //初始化线程二
        Thread t2=new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    t1.join();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    System.out.println("t2 is running...");
                }
            }
        });

        //初始化线程三
        Thread t3=new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    t2.join();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    System.out.println("t3 is running...");
                }
            }
        });

        t1.start();
        t2.start();
        t3.start();
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

输出结果

t1 is running...
t2 is running...
t3 is running...
1
2
3

Thread.join()方法部分源码

public final synchronized void join(long millis) throws InterruptedException {
   //条件不足继续等待
    while (isAlive()) {
        wait(0);
    }
    //条件符合，方法返回
}
1
2
3
4
5
6
7

ThreadLocal 的使用

ThreadLocal 即线程变量，以一个 threadLocal 对象为键，任意对象为值的存储结构。这个结构被附带在线程上，也就是一个线程可以根据一个 threadLocal 对象查询到绑定在这个线程上的一个值。可通过 set(T)方法来设置一个值，在当前线程下再通过 get() 方法获取到原先设置的值。

可以用这个实现一个工具类来用aop得到一个方法的执行时间

等待超时模式

开发人员常常会遇到：调用一个方法时等待一段时间，如果该方法能在给定时间段内得到结果，那么结果立刻返回，反之，超时返回默认结果。

这为伪代码

public synchronized Object get(long mills) throws InterruptedException{
        long future = System.currentTimeMillis()+mills;
        long remaining = mills;
//        执行一段任务获得 result并且 result不满足要求
        while (result==null&&remaining>0){
            wait(remaining);
            remaining = future-System.currentTimeMillis();
        }
        return result;
    }
1
2
3
4
5
6
7
8
9
10

可以看出，等待超时模式就是在等待/通知范式上增加了超时控制，这使得该模式比原有范式更有灵活性，因为即使该方法执行时间过长，也不会“永久”阻塞调用者，而是会按照调用者的要求“按时”返回。

java 中的锁

AQS 同步器

AQS 队列同步器详解

ReentrantLock重入锁

重进入是指任意线程在获取到锁之后，能够再次获取该锁而不会被阻塞，该特性实现需要解决两个问题

1. 线程再次获取锁。锁需要识别锁的线程是否为当前占据锁的线程，如果是则再次成功获取。
2. 锁的最终释放。线程重复n次获取锁，随后在第n次释放锁后，其他线程能够获取到该锁。锁的最终释放要求锁对于获取进行计数自增，计数表示当前锁被重复获取的次数，锁被释放计数自减，当计数为0表示锁已成功获取。

ReentrantLock 是通过组合自定义同步器来实现锁的获取与释放，以非公平性的实现为例.

final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

该方法通过判断当前线程是否为获取锁的线程来决定获取操作是否成功，如果是获取锁的线程再次请求，则将同步状态值增加并返回 true，表示获取同步状态成功。

成功获取锁的线程再获取锁只是增加了同步状态值，也要求了 ReentLock 在释放同步状态时减少同步状态值。

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;
}
1
2
3
4
5
6
7
8
9
10
11
12

公平性与否是针对获取锁而言，如果一个锁是公平的，那么一个锁的获取顺序就应该符合请求的绝对时间顺序。

//reentrantLock的公平锁获取
protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) {//唯一不同是这里加了 hasQueuedPredecessors 方法如果返回 true，则表示有线程比当前线程更早地请求锁，因此需要等待前驱线程获取并释放锁之后才能继续获取
                    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;
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

公平锁与非公平锁相比保证了锁的获取原则按照 FIFO 原则，而代价是进行大量的线程切换，非公平锁虽然可能造成线程的“饥饿”，但极少线程切换，保证了更大的吞吐量。

读写锁

ReentrantReadWriteLock特性

ReentrantRedWriteLock的接口

写锁的获取与释放

写锁的获取

protected final boolean tryAcquire(int acquires) {
    Thread current = Thread.currentThread();
    int c = getState();//这里的c应该是写锁获取的次数
    int w = exclusiveCount(c);//读锁和写锁不能同时被两个线程持有
    if (c != 0) {
        // 存在读锁或者当前获取线程已经不是获取写锁的线程
        if (w == 0 || current != getExclusiveOwnerThread())
            return false;
        if (w + exclusiveCount(acquires) > MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // Reentrant acquire
        setState(c + acquires);
        return true;
    }
    if (writerShouldBlock() ||
        !compareAndSetState(c, c + acquires))
        return false;
    setExclusiveOwnerThread(current);
    return true;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

该方法重了重入条件之外，增加一个读锁是否存在的判断。如果存在读锁，则写锁不能被获取，原因在于：读写锁要确保写锁的操作对于读锁可见，如果允许读锁在被已经获取的情况下对写锁的获取，那么正在运行的其它线程就无法感知当前写线程的操作。

写锁的释放

它的释放与 ReentrantLock 的释放过程基本类似，每次释放均减少写状态，当写状态为 0 时表示写锁已被释放，从而等待的读写线程能够继续访问读写锁，同时前次写线程的修改对后续线程可见。

读锁的获取与释放

读锁是一个支持重进入的共享锁，它能同时被多个线程同时获取，在没有其他线程访问时，读锁总会被成功地获取，所做的也啥师增加读状态。

这里将获取读锁的代码做了删减，保留必要的部分。

protected final int tryAcquiredsShared(int unused){
    for(;;){
        int c = getState();
        int nextc = c + (1 << 16);
        if(nextc<c)
            throw new Error("Maximum lock count exceeded");
        if(exclusiveCount(c) != c && owner != Thread.currentThread()){
            return -1;
        }
        if(compareAndSetState(c,nextc))
            return 1;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

在此方法中，如果其他线程已经获取了写锁，则当前线程获取读锁失败，进入等待状态。如果当前线程获取了写锁或者写锁未被获取，则当前线程增加读状态，成功获取读锁。

LockSupport 工具

当需要阻塞或者唤醒一个线程的时候，都会使用LockSupport工具类来完成相应的操作。

Condition 接口

任意一个 java 接口，都拥有一组监视器方法(定义在 java.lang.Object上)，主要包括 wait()、wait(long timeout)、notify以及 notifyAll()方法，这些方法与 synchronized 同步关键字配合可以实现等待/通知模式。Condition 接口也提供了类似 Object 的监视器器方法,与 Lock 配合可以实现等待/通知模式,但这两回合在使用方式以及功能特性上还是有区别。

Condition 接口与示例

当前线程调用Conditiony方法时，需要提前获取到 Condition 对象关联的锁。Condtion对象是由 Lock 对象(调用Lock对象的 newCondition方法)创建出来的。

示例

	static ReentrantLock lock;
	static Condition condition;

    public static void main(String[] args) throws InterruptedException {
        lock = new ReentrantLock();
        condition = lock.newCondition();
    }
    public void conditionWait() throws InterruptedException{
        lock.lock();
        try{
            condition.await();
        }finally {
            lock.unlock();
        }
    }

    

    public static void conditionSignal() throws InterruptedException{
        lock.lock();
        try{
            condition.signal();
        }finally {
           lock.unlock(); 
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

接口

Condition的实现分析

Condition 是同步器 AbstractQueuedSynchronizer 的内部类。每个 Condition 对象都包含着一个队列，该队列是 Condition 对象实现等待/通知功能的关键。

下面分析其实现，主要包括：等待队列、等待和通知。

1. 等待队列

​ 等待队列是一个 FIFO 的队列，在队列中的每一个节点都包含一个线程引用，该线程就是在 Condition 对象上等待的线程，如果一个线程调用了 Condition.await() 方法，那么该线程将会释放锁、构成节点加入等待队列并进入等待状态。

​ 一个Condition 包含一个等待队列， Conditon 拥有首节点(firstWaiter)与尾节点(lastWaiter)。当前线程调用 Condition.await() 方法，将会以当前线程构造节点，并将节点从尾部加入等待队列。

Condition 拥有首尾节点的引用，而新增节点只需要将原有尾节点 nextWaiter指向它，并更新尾节点即可。上述操作用锁来保证线程的安全。

在Object 的监视器模型上，一个对象拥有一个同步队列与等待队列，而并发包中的 Lock 拥有一个同步队列与多个等待队列。

2. 等待

​ 调用 Condition 的 await() 方法，会使当前线程进入等待队列并释放锁，同时线程状态变为等待状态。

如果从队列(同步队列与等待队列)的角度来看 await(方法)，当调用了 await() 方法时，相当于同步队列的首节点(获取锁的节点)移动到 Condition 的等待队列中。

ConditionObject 的 await 方法

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addConditionWaiter();//当前线程加入等待队列
    int savedState = fullyRelease(node);//释放同步状态，也就是释放锁
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

调用该方法的线程成功获取了锁的线程，也就是同步队列的首节点，该方法会将当前线程构成节点并加入等待队列中，然后释放同步状态，唤醒同步队列中的后继节点，然后当前线程会进入等待状态。

当等待队列中的节点被唤醒，则唤醒节点的线程开始尝试获取同步状态。如果不是通过其他线程调用 Condition.signal() 方法唤醒，而是对等待线程进行中断则会抛出 InterruptedException。

3. 通知

调用 Condition的 Signal() 将会唤醒首节点在唤醒之前，会将节点移入同步队列

public final void signal() {
            if (!isHeldExclusively())//确认当前线程必须是获取了锁的线程
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;//首节点
            if (first != null)
                doSignal(first);//将其移动到同步队列并使用 LockSupport 唤醒节点中的线程
}
1
2
3
4
5
6
7

被唤醒的线程，将从 await() 方法中的 while 循环中退出，进而调用同步器的 acquiredQueued() 方法加入到获取同步状态的竞争中。

成功获取同步状态(或者说锁)，被唤醒的线程将从先前调用的 await() 方法返回，此时线程已经成功获取了锁

java并发容器和框架

ConcurrentHashMap 的实现原理与使用

HashTable 使用 sychronized 来保证线程安全只有一把锁，当多线程并发时只有一个线程能操作。HashMap线程不安全，在多线程并发情况下执行 put 操作会引起死循环，因为多线程会导致HashMap的 Entry 链表形成环形结构，一但形成环形 Entry 的 next 的节点永远不为空，就会产生死循环获取 Entry。

而 ConcurrentHashMap 使用锁分段技术，首先将数据分成一段一段地存储，然后给每个数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其它段的数据也能访问。

java.util.concurrent.ConcurrentHashMap属于 JUC 包下的一个集合类，可以实现线程安全。

它由多个 Segment 组合而成。Segment 本身就相当于一个 HashMap 对象。同 HashMap 一样，Segment 包含一个 HashEntry 数组，数组中的每一个 HashEntry 既是一个键值对，也是一个链表的头节点。

单一的 Segment 结构如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ayOeJ3N9-1661604160130)(D:\notebook\文档截图\Segment结构.png)]

static class Segment<K,V> extends ReentrantLock implements Serializable {
    private static final long serialVersionUID = 2249069246763182397L;
    final float loadFactor;
    Segment(float lf) { this.loadFactor = lf; }
}
1
2
3
4
5

像这样的 Segment 对象，在 ConcurrentHashMap 集合中有多少个呢？有 2 的 N 次方个，共同保存在一个名为 segments 的数组当中。

因此整个ConcurrentHashMap的结构如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-RX0mhrIx-1661604160132)(D:\notebook\文档截图\ConcurrentHashMap结构.png)]

可以说，ConcurrentHashMap 是一个二级哈希表。在一个总的哈希表下面，有若干个子哈希表。

这些理论知识好像没有什么用,我所看到的 Segment 对象并没有被大范围的使用,反而基本没有用到,但确实找到了它作为一个内部类出现在 ConcurrentHashMap 在整个类里默默无闻

前言

在JDK1.7中的ConcurrentHashMap中我们了解到ConcurrentHashMap采用分段锁的机制，实现并发的更新操作，它首先将数据分成一段一段地存储，然后给每一段数据配一个锁，当一个线程占用锁访问其中一段数据时，其他段的数据也能被其他线程访问，锁分段技术的使用大大了提高并发访问效率。底层由ReentrantLock+Segment+HashEntry组成。然而在jdk1.8中的实现已经抛弃了Segment分段锁机制，利用CAS+Synchronized来保证并发更新的安全，底层依然采用数组+链表+红黑树的存储结构。

问题1：为什么在1.8中舍弃了分段锁的机制？

其实可以看出JDK1.8版本的ConcurrentHashMap的数据结构已经接近HashMap，相对而言，ConcurrentHashMap只是增加了同步的操作来控制并发，从JDK1.7版本的ReentrantLock+Segment+HashEntry，到JDK1.8版本中synchronized+CAS+HashEntry+红黑树,相对而言，总结如下思考

• JDK1.8的实现降低锁的粒度，JDK1.7版本锁的粒度是基于Segment的，包含多个HashEntry，而JDK1.8锁的粒度就是HashEntry（首节点）

• JDK1.8版本的数据结构变得更加简单，使得操作也更加清晰流畅，因为已经使用synchronized来进行同步，所以不需要分段锁的概念，也就不需要Segment这种数据结构了，由于粒度的降低，实现的复杂度也增加了

• JDK1.8使用红黑树来优化链表，基于长度很长的链表的遍历是一个很漫长的过程，而红黑树的遍历效率是很快的，代替一定阈值的链表，这样形成一个最佳拍档

• JDK1.8为什么使用内置锁synchronized来代替重入锁ReentrantLock，我觉得有以下几点

  1. 因为粒度降低了，在相对而言的低粒度加锁方式，synchronized并不比ReentrantLock差，在粗粒度加锁中ReentrantLock可能通过Condition来控制各个低粒度的边界，更加的灵活，而在低粒度中，Condition的优势就没有了
  2. JVM的开发团队从来都没有放弃synchronized，而且基于JVM的synchronized优化空间更大，使用内嵌的关键字比使用API更加自然
  3. 在大量的数据操作下，对于JVM的内存压力，基于API的ReentrantLock会开销更多的内存，虽然不是瓶颈，但是也是一个选择依据。
     转自：https://blog.csdn.net/q289336929/article/details/95742247

我们通过了解1.8的HashMap知道在插入新节点后，插入新节点所在的红黑树的根节点可能发生改变。我们假设在1.8的ConcurrentHashMap也与HashMap一样，那么首先会以该红黑树的根节点为对象加锁，那么有可能在插入新节点过后该红黑树的根节点发生改变，这样就会造成第二个线程同时在操作该红黑树的时候，原本线程未结束因为根节点对象改变导致线程2获取到了锁，从而导致产生数据冲突这样的一个问题。而TreeBin对象将整棵树包起来，从而对TreeBin作为锁对象，就不会因根节点改变而导致上述问题。

ConcurrentHashMap的操作

1. get 操作

get操作的高效之处在于整个 get 过程都不需要加锁。我们知道 HashTable 的 get 操作是需要加锁的，那么ConcurrentHashMap 是如何做到不加锁的呢？它的 get 方法里将要使用的共享变量都定义成 volatile 类型，当前 hash 值与 key 值都设置为 final 也杜绝了修改用于存储当前节点值的 val 与 下一个节点的 next。定义成 volatitle 的变量，能够在线程之间保持可见性，能够被多线程读，并且保证不会读到过期的值，但是只能被单线程写。由于这里不需要写，所以不用加锁。之所以不会读到过期值，是因为 java 内存模型的 happen before 原则，对 volatile 字段的写入操作等于读操作，即使两个线程同时修改和获取 volatile 变量，get 操作也能拿到最新的值。

static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    volatile V val;
    volatile Node<K,V> next;
 }
1
2
3
4
5
6

2. put 操作

final V putVal(K key, V value, boolean onlyIfAbsent) {
        //如果key或value为空则抛空指针异常
        if (key == null || value == null) throw new NullPointerException();
    //用 key 得到应该存放位置的 hash 值
        int hash = spread(key.hashCode());
        int binCount = 0;
        //遍历数组
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
            //如果数组为空，则初始化一个数组
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
            //如果table[index]位置为空，则创建Node对象插入其中
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                if (casTabAt(tab, i, null,
                             new Node<K,V>(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            }
            //如果key位置的hash值为-1（MOVED），标志这其他线程在对数组进行扩容，本线程帮助其他线程一起扩容，加快效率。
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
           //key位置上不为空，存在链表或红黑树或单个Node对象
            else {
                V oldVal = null;
                //对要加入的Node对象上锁，在1.8中优化了synchronized ，所以效率没有以前那么低
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                       //如果是链表
                        if (fh >= 0) {
                            binCount = 1;
                            //遍历该链表
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                //位置冲突则进行值覆盖，返回旧值
                                if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                Node<K,V> pred = e;
                                //尾插法，插入到链表尾部
                                if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,
                                                              value, null);
                                    break;
                                }
                            }
                        }
                       //如果是红黑树
                        else if (f instanceof TreeBin) {
                            Node<K,V> p;
                            binCount = 2;
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                           value)) != null) {
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }
                if (binCount != 0) {
                    //判断链表长度是否大于等于8，是则转为红黑树
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    //返回旧值
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        //数组中元素个数加一
        addCount(1L, binCount);
        return null;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

3. size 操作

size方法最后返回 baseCount属性 加上CounterCell数组里面的所有值的和。

public int size() {
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
    }
   
//返回
final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

CounterCell是一个静态内部类，里面的long属性是通过volatile 修饰，来保证并发安全。

@sun.misc.Contended static final class CounterCell {
        volatile long value;
        CounterCell(long x) { value = x; }
    }
 /**
   * Table of counter cells. When non-null, size is a power of 2.
   * 计数器单元表。当非空时，大小是2的乘方。
  */
private transient volatile CounterCell[] counterCells;
1
2
3
4
5
6
7
8
9

4. addCount

private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
    if ((as = counterCells) != null ||//如果为空则对当前 map 对象 cas 操作baseCount + 1 cas 成功就跳过 
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        CounterCell a; long v; int m;
        boolean uncontended = true;
        if (as == null || (m = as.length - 1) < 0 //如果不为空就通过当前线程的 hash 值在此线程对应的位置，
            (a = as[ThreadLocalRandom.getProbe() & m]) == null ||//如果这个位置为空就执行 fullAddCount
            !(uncontended =//如果不为空就对此CounterCell对象cas操作value加1。如果成功return;失败就执行fullAddCount方法
              U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
            fullAddCount(x, uncontended);//这里执行
            return;
        }
        if (check <= 1)
            return;
        s = sumCount();
    }
    if (check >= 0) {//这里的代码是不竞争锁的情况？
        Node<K,V>[] tab, nt; int n, sc;
        while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
               (n = tab.length) < MAXIMUM_CAPACITY) {
            int rs = resizeStamp(n);
            if (sc < 0) {
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                    transferIndex <= 0)
                    break;
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

大概意思就是多个线程去调用put方法，也就是多个线程去size+1。在ConcurrentHashMap中通过baseCount去计数。在高并发的场景下，通过CAS机制去控制baseCount+1，也就是只有一个线程能够操作成功。其他线程通过 “随机数&table.length-1” 获取到CounterCell的数组下标，然后去操作CounterCell数组对应下标对象中的value属性，使其value属性+1。最后统计baseCount+CounterCell的数目。

为什么不直接 for 循环对当前 map 对象 cas 操作 baseCount 加 1，却要引入CountCell数组

因为for循环cas这种方式可以解决多线程并发问题，但因为cas的是当前map对象，所以同一时刻还是只有一个线程能cas成功，而对于引入CounterCell数组，cas的是当前线程对应在数组中特定位置的元素，也就是说如果位置不冲突，n个长度的CounterCell数组是可以支持n个线程同时cas成功的。

总结：以数组的形式去分散线程，防止多个线程去操作属性只有一个线程能够成功，从而导致浪费其他线程的资源，进而提高并发效率。

ConcurrentLinkedQueue

1. 入队

public boolean offer(E var1) {
        checkNotNull(var1);
        ConcurrentLinkedQueue.Node var2 = new ConcurrentLinkedQueue.Node(var1);
        ConcurrentLinkedQueue.Node var3 = this.tail;
        ConcurrentLinkedQueue.Node var4 = var3;

        do {
            while(true) {
                ConcurrentLinkedQueue.Node var5 = var4.next;
                //获得下一个节点
                if (var5 == null) {//如果下一个节点为空就出循环
                    break;
                }

                if (var4 == var5) {
                    var4 = var3 != (var3 = this.tail) ? var3 : this.head;
                } else {//这两个代码主要是将这几个变量一直向后移动
                    var4 = var4 != var3 && var3 != (var3 = this.tail) ? var3 : var5;
                }
            }
        } while(!var4.casNext((ConcurrentLinkedQueue.Node)null, var2));//即找到一个可插入的下一个节点为空的节点,并将 var2 插入
        if (var4 != var3) {
            this.casTail(var3, var2);//更新tail节点，允许失败
        }

        return true;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

2. 出队

public E poll() {
    while(true) {
        ConcurrentLinkedQueue.Node var1 = this.head;
        ConcurrentLinkedQueue.Node var2 = var1;

        while(true) {
            Object var4 = var2.item;
            ConcurrentLinkedQueue.Node var3;
            //如果 var4不等于空就将 var2的item设为 var4
            if (var4 != null && var2.casItem(var4, (Object)null)) {
                if (var2 != var1) {//将头结点的下一个结点设为头结点如果没有线程修改 var1 与 var3
                    this.updateHead(var1, (var3 = var2.next) != null ? var3 : var2);
                }

                return var4;//将头结点的值返回
            }

            if ((var3 = var2.next) == null) {//到这 head.item为null 说明就1结点直接将头结点中的值设置为null
                this.updateHead(var1, var2);
                return null;
            }

            if (var2 == var3) {
                break;
            }

            var2 = var3;
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

阻塞队列

1. 什么是阻塞队列？

阻塞队列（BlockingQueue） 是一个支持两个附加操作的队列。这两个附加的操作是：在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程会等待队列可用。阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

七种阻塞队列

ArrayBlockingQueue
数组实现的有界阻塞队列，此队列按照先进先出FIFO原则对元素进行排序

LinkedBlockingQueue
链表实现的有界阻塞队列，此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则

PriorityBlockingQueue
支持优先级排序的无界阻塞队列，默认情况下元素采用自然排序升序排列。也可以自定义类实现compareTo()方法来指定元素排序规则，或者初始化PriorityBlockingQueue时，指定构造参数Comparetor来对元素进行排序

DelayQueue
优先级队列实现的无界阻塞队列

SynchronousQueue
不存储元素的阻塞队列，每一个put操作必须必须等待一个tack操作，否则不能继续添加元素

LinkedTransferQueue
链表实现的无界阻塞队列

LinkedBlockingDeque
链表实现的双向阻塞队列

阻塞队列的实现原理

使用通知模式实现。所谓通知模式就是生产者往满队列里添加元素时会阻塞住生产者，当消费者消费了一个队列中的元素后，会通知生产者当前队列可用。

这里拿 ArrayBlockingQueue 的 put 操作举例

 public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;//这里使用锁来保证线程安全
        lock.lockInterruptibly();
     //这里使用lockInterruptibly 方法目的是防止线程等待，让等待的线程做些其它事
        try {
            while (count == items.length)//如果队列满了就将线程移入 notnull 的等待队列中
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
}
private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[putIndex] = x;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        notEmpty.signal();
    //唤醒消费者来消费
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

fork/join 框架

一、fork/join定义

fork/join框架思想：简单来说就是将工作拆分成最小单位不可拆分级别，然后在各自相加连接 ，取得结果。
递归分合，分而治之。

工作窃取模式

它是指某个线程从其他队列里窃取任务来执行。而这时它们会访问同一个队列，通常会使用双端队列，被窃取任务的线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。

• 优点：充分利用线程进行并行计算，减少了线程之间的竞争
• 缺点：在某些情况下还是存在竞争，比如双端队列中只有一个任务时。该算法会消耗更多的资源，比如创建多个线程与双端队列。

框架实现原理

在Java的Fork/Join框架中，它提供了两个类来帮助我们完成任务分割以及执行任务并合并结果：

1. ForkJoinTask：我们要使用ForkJoin框架，必须首先创建一个 ForkJoin 任务。它提供在任务中执行 fork() 和join()操作的机制，通常情况下我们不需要直接继承ForkJoinTask类，而只需要继承它的子类，Fork/Join框架提供了以下两个子类：

   RecursiveAction：用于没有返回结果的任务。
   RecursiveTask ：用于有返回结果的任务。

2. ForkJoinPool ：ForkJoinTask需要通过ForkJoinPool来执行，任务分割出的子任务会添加到当前工作线程所维护的双端队列中，进入队列的头部。当一个工作线程的队列里暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务。

ForkJoinPool里面，有两个特别重要的成员如下：

volatile WorkQueue[] workQueues;     // main registry
final ForkJoinWorkerThreadFactory factory;
1
2

workQueues 用于保存向ForkJoinPool提交的任务，而具体的执行由ForkJoinWorkerThread执行，而ForkJoinWorkerThreadFactory可以用于生产出ForkJoinWorkerThread：

public static interface ForkJoinWorkerThreadFactory {
    /**
        * Returns a new worker thread operating in the given pool.
        *
        * @param pool the pool this thread works in
        * @return the new worker thread
        * @throws NullPointerException if the pool is null
        */
    public ForkJoinWorkerThread newThread(ForkJoinPool pool);
}
1
2
3
4
5
6
7
8
9
10

ForkJoinPool里面，有两个特别重要的成员如下：

volatile WorkQueue[] workQueues;     // main registry
final ForkJoinWorkerThreadFactory factory;
1
2

workQueues 用于保存向ForkJoinPool提交的任务，而具体的执行由 ForkJoinWorkerThread 执行，而 ForkJoinWorkerThreadFactory 可以用于生产出 ForkJoinWorkerThread ：

public static interface ForkJoinWorkerThreadFactory {
    /**
        * Returns a new worker thread operating in the given pool.
        *
        * @param pool the pool this thread works in
        * @return the new worker thread
        * @throws NullPointerException if the pool is null
        */
    public ForkJoinWorkerThread newThread(ForkJoinPool pool);
}
1
2
3
4
5
6
7
8
9
10

ForkJoinTask的fork方法实现原理

public final ForkJoinTask<V> fork() {
    Thread t;
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        //如果当前线程是 forkJoinWorkerThread 线程，则向下转型成 ForkJoinWorkerThread ,并将当前任务 Push 到当前线程负责的队列中
        ((ForkJoinWorkerThread)t).workQueue.push(this);
    else
        ForkJoinPool.common.externalPush(this);
    // 要不然就放入 WorkerPool 的工作队列中
    return this;
}
1
2
3
4
5
6
7
8
9
10

在ForkJoinWorkerThread类中有一个pool和一个workQueue字段：

// 线程工作的ForkJoinPool
final ForkJoinPool pool;                // the pool this thread works in
// 工作窃取队列
final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics
1
2
3
4

workQueue的push()方法如下：

final void push(ForkJoinTask<?> task) {
    ForkJoinTask<?>[] a; ForkJoinPool p;
    int b = base, s = top, n;
    if ((a = array) != null) {    // ignore if queue removed
        int m = a.length - 1;     // fenced write for task visibility
        U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);
        U.putOrderedInt(this, QTOP, s + 1);
        //调用 Unsafe 类将 task 放入内存
        if ((n = s - b) <= 1) {//如果还有 workerThread 就将任务放入工作队列，并唤醒线程
            if ((p = pool) != null)
                p.signalWork(p.workQueues, this);
        }
        else if (n >= m)
            growArray();
    }
}

volatile int base;         // index of next slot for poll
int top;                   // index of next slot for push
ForkJoinTask<?>[] array;   // the elements (initially unallocated)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

这里普及一下 Unsafe 类

1.Unsafe类介绍

Unsafe类是在sun.misc包下，不属于Java标准。但是很多Java的基础类库，包括一些被广泛使用的高性能开发库都是基于Unsafe类开发的，比如Netty、Hadoop、Kafka等。

使用Unsafe可用来直接访问系统内存资源并进行自主管理，Unsafe类在提升Java运行效率，增强Java语言底层操作能力方面起了很大的作用。

Unsafe可认为是Java中留下的后门，提供了一些低层次操作，如直接内存访问、线程调度等。

官方并不建议使用Unsafe。

更多的推荐看这篇博客 https://www.jb51.net/article/140726.htm

ForkJoinTask的join方法实现原理

join 方法的主要作用是阻塞当前线程并等待获取结果，其源码如下：

public final V join() {
    int s;
    if ((s = doJoin() & DONE_MASK) != NORMAL)
        reportException(s);
    return getRawResult();
}
private void reportException(int s) {//就是个根据状态抛异常的方法
    if (s == CANCELLED)
        throw new CancellationException();
    if (s == EXCEPTIONAL)
        rethrow(getThrowableException());
}
private int doJoin() {
    int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
    return (s = status) < 0 ? s ://查看是否执行完成 
        ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?//如果没有执行完成，就查看当前线程是否是
        (w = (wt = (ForkJoinWorkerThread)t).workQueue).//ForkJoinWorkerThread 如果是就将任务放入它的工作队列并执行
        tryUnpush(this) && (s = doExec()) < 0 ? s ://如果执行的结果完成了就直接返回，如果没有完成就等待其它线程来一起完成 
        wt.pool.awaitJoin(w, this, 0L) :
        externalAwaitDone();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

它调用了doJoin()方法，通过doJoin()方法得到当前任务的状态来判断返回什么结果，任务状态有四种：

• 已完成（NORMAL),直接返回结果
• 被取消（CANCELLED），被取消抛出 CancellationException
• 信号（SIGNAL）
• 出现异常（EXCEPTIONAL）： 直接抛出对应的异常

java 中的 13 个原子类

原子更新基本类型

• AtomicBoolean:原子更新布尔类型
• AtomicInteger:原子更新整型
• AtomicLong:原子更新长整型

三个类几乎一模一样，所以这里以 AtomicInterger 为例。

常用方法如下：

								AtomicInterger.java
public final int getAndIncrement() {
    return unsafe.getAndAddInt(this, valueOffset, 1);
}
								Unsafe.class
public final int getAndAddInt(Object var1, long var2, int var4) {
        int var5;
        do {
            var5 = this.getIntVolatile(var1, var2);//从内存中得到这个值
        } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));//如果是这个值就将这个值加上1，如果不成功就一直自旋

        return var5;
    }
public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);//方法为native方法也就是本地方法，用c或者c++写的。
1
2
3
4
5
6
7
8
9
10
11
12
13
14

原子更新数组

通过原子方式更新数组里某个元素，Atomic 包提供了以下 3 个类。

• AtomicIntegerArray: 原子更新整形数组里的元素
• AtomicLongArray: 原子更新长整形数组里的元素
• AtomicReferenceArray: 原子更新引用类型数组里的元素
• AtomicDoubleArray: 原子更新浮点数组里的元素

这里以 AtomicIntegerArray 为示范,常用方法如下

原子更新引用类

原子更新基本类型 AtomicInteger,只能更新一个变量，如果要更新多个变量，就需要使用这个原子更新引用类型提供的类。

• AtomicReference 原子更新引用类
• AtomicReferenceFieldUpdater 原子更新引用类型里的字段
• AtomicMarkableReference 原子更新带有标记位的引用类型。可以原子更新一个布尔类型的标记位和引用类型。

public class AtomicReferenceTest {
    public static AtomicReference<user> atomicUserRef = new
        AtomicReference<user>();
    public static void main(String[] args) {
        User user = new User("conan"， 15);
        atomicUserRef.set(user);
        User updateUser = new User("Shinichi"， 17); 
        //用CAS的方式更新引用
        atomicUserRef.compareAndSet(user， updateUser);
        System.out.println(atomicUserRef.get().getName());
        System.out.println(atomicUserRef.get().getOld());
    }
    static class User {
        private String name;
        private int old;
        public User(String name， int old) {
            this.name = name;
            this.old = old;
        }
        public String getName() {
            return name;
        }
        public int getOld() {
            return old;
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

原子更新字段类

如果需要原子地更新某个类里的某个字段时，就需要使用原子更新字段类，Atomic包提供了以下3个类进行原子字段更新

• AtomicIntegerFieldUpdater：原子更新整型的字段的更新器
• AtomicLongFieldUpdater：原子更新长整型字段的更新器
• AtomicStampedReference：原子更新带有版本号的引用类型。该类将整数值与引 用关联起来，可用于原子的更新数据和数据的版本号，可以解决使用 CAS 进行 原子更新时可能出现的 ABA 问题

public class AtomicIntegerFieldUpdaterTest {
    // 创建原子更新器，并设置需要更新的对象类和对象的属性
    //由于AtomicIntegerFieldUpdater是一个抽象类，所以使用的时候需要使用静态方法：newUpadter()创建一个更新器，并且设置想要更新的类和属性；并且更新类的字段（属性）必须使用 public volatile 修饰符
    private static AtomicIntegerFieldUpdater<User> a = AtomicIntegerFieldUpdater.
        newUpdater(User.class， "old");
    public static void main(String[] args) { // 设置柯南的年龄是 10 岁
        User conan = new User("conan"， 10);
        // 柯南长了一岁，但是仍然会输出旧的年龄 ===》输出：10
        System.out.println(a.getAndIncrement(conan));
        // 输出柯南现在的年龄 ===》输出：11
        System.out.println(a.get(conan));
    }
    public static class User {
        private String name;
        public volatile int old;
        public User(String name， int old) {
            this.name = name;
            this.old = old;
        }
        public String getName() {
            return name;
        }
        public int getOld() {
            return old;
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

java 中的并发工具类

在JDK的并发包里提供了几个非常有用的并发工具类。CountDownLatch、CyclicBarrier和Semaphore工具类提供了一种并发流程控制的手段，Exchanger工具类提供了在线程间交换数据的一种方法。本章会配合一些应用场景来介绍如何使用这些工具类。

CountDownLatch

等待多线程完成的CountDownLatch

CountDownLatch允许一个或多个线程等待其他线程完成操作

示例

public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(2);
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println(1);
                countDownLatch.countDown();
                System.out.println(2);
                countDownLatch.countDown();
            }
        }).start();
        countDownLatch.await();
        System.out.println(3);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

CountDownLatch的构造函数接收一个 int 类型的参数作为计算器，如果你想等待 N 个点完成，这里就传入N。
当我们调用CountDownLatch的countDown方法时，N就减1，CountDownLatch的await方法会阻塞当前线程，指定N变成零。由于countDown方法可以用在任何地方，所有这里说的N个点，可以是N个线程，也可以是1个线程的N个执行步骤。用在多个线程时，只需要把这个CountDownLatch的引用传递到线程里即可。
如果某个线程处理的比较慢，我们不可能让主线程一直等待，所以可以使用另外一个带指定时间的await方法一一await(long time, TimeUnit unit)，这个方法等待特定时间后，就会不再阻挡当前线程。join也有类似的方法。

同步屏障 CyclicBarrier

CyclicBarrier的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。

CyclicBarrier简介
CyclicBarrier默认的构造方法是CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。示例代码如下：

public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(2);
        new Thread(() -> {
            try {
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println(1);
        }).start();
        try {
            cyclicBarrier.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println(2);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

因为主线程和子线程的调度是有CPU决定的，两个线程都有可能先执行，所以会产生两种输出，第一种可能是1,2；第二种是2,1。
如果把new CyclicBarrier(2)改成CyclicBarrier(3)，则主线程和子线程会永远等待，因为没有第三个线程执行await方法，即没有第三个线程到达屏障，所以之前到达屏障的两个线程都不会继续执行。

CyclicBarrier还提供了一个高级的构造函数CyclicBarrier(int parties, Runnable barrierAction)，用于在所有线程都到达屏障时，优先执行barrierAction，CyclicBarrier使用BarrierAction只是说明所有线程已经到达屏障了，然后CyclicBarrier执行barrierAction任务，最后线程在各自继续执行。这个构造方法方便处理更复杂的业务场景。

 public static void main(String[] args) {
        CyclicBarrier c = new CyclicBarrier(2, new A());
        new Thread(() -> {
            try {
                c.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(2);
        }).start();
        try {
            c.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println(1);
    }

    static class A implements Runnable {

        @Override
        public void run() {
            System.out.println(3);//会优先输出 
        }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

CyclicBarrier和CountDownLatch的区别
使用的场景不一样，CyclicBarrier阻塞每个线程（每个线程都等待最后一个线程到达屏障），像分布式问题，屏障清除了，每个线程可以继续做自己的事情；CountDownLatch阻塞等待线程（等待线程等待其他线程完成）向集中式的，最后由等待线程统一处理。
CountDownLatch的计数器只能使用一次，而CyclicBarrier的计数器可以使用reset()方法重置。所以CyclicBarrier能处理更为复杂的业务场景。例如，如果计算发生错误，可以重置计数器，并让线程重弄下执行一次。
CyclicBarrier还提供其他有用的方法，比如getNumberWaiting方法可以获得CyclicBarrier阻塞的线程数量。isBroken()方法用来了解阻塞的线程是否被中断。

控制并发线程数的Semaphore

Semaphore（信号量）是用来控制同时访问特定资源的线程数量，它通过协调各个线程，以保证合理的使用公共资源。

仅仅从字面上很难理解Semaphore所表达的含义，可以把它比作停车场控制流量的指示灯。比如停车场的容量是1000，只允许同时有一千辆车在这个停车场，其他的都必须在入口等待，所以前一千辆车会看到绿灯，可进入停车场，当停车场停满后指示灯变红，其他车不能进入停车场。但是如果前一千辆车有10辆已经离开了停车场，那么后面的10辆车就允许进入停车场。这个例子里说的车就是线程，进入停车场表示线程在执行，离开停车场表示线程执行完成，看见红灯表示线程被阻塞，不能执行。

应用场景
Semaphore可以做流量控制，特别是公用资源有限的应用场景，比如数据库连接。
假如有一个需求，要读取几万个文件的数据，因为都是IO密集型任务，我们可以启动几十个线程并发的读取，但是如果读到内存后，还需要存储到数据库中，而数据库连接数只有10个，这是我们必须控制只有10个线程同时获取数据库连接保持数据，否则会报错无法获取数据库连接。这个时候，就可以使用Semaphore来做流量控制。

private static final int THREAD_COUNT = 30;
private static final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
private static final Semaphore s = new Semaphore(10);
    public static void main(String[] args) {
        for (int i = 0; i < THREAD_COUNT; i++) {
            executorService.execute(() -> {
                try {
                    s.acquire();
                    System.out.println("save data");
                    s.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

Semaphore的构造方法Semaphore(int permits)接收一个整型的数组，表示可用的许可证数量。Semaphore(10)表示允许10个线程获取许可证，也就是最大并发数是10。Semaphore的用法也很简单，首先线程使用Semaphore的acquire()方法获取一个许可证，使用完之后调用release()方法归还许可证。还可以使用tryAcquire()方法尝试获取许可证。

其它方法

线程间交换数据的 Exchanger

Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。这两个线程通过exhange()方法交换数据，如果第一个线程先执行exhange()方法，它会一直等待第二个线程也执行exchange方法，当两个线程都达到同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。

示例

Exchange可以用于校对工作，比如我们需要将纸质银行流水通过人工的方式录入成电子银行流水，为了避免错误，采用AB岗两人进行录入，录入到Excel之后，系统需要加载这两个Excel，并对两个Excel数据进行校对，看看是否录入一致，代码如下：

private static final Exchanger<String> exgr = new Exchanger<>();
    private static final ExecutorService threadPool = Executors.newFixedThreadPool(2);
    public static void main(String[] args) {
        threadPool.execute(() -> {
            String a = "银行流水A";
            String exchange = null;
            try {
                exchange = exgr.exchange(a);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(exchange);
        });

        threadPool.execute(() -> {
            String b = "银行流水B";
            try {
                String exchange = exgr.exchange(b);
                System.out.println("A 录入的是:" + exchange + " B录入的是：" + b);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

如果两个线程一个没有执行exchange（）方法，则会一直等待，如果单向特殊情况发生，避免一直等待，可以使用exchange(V x, long timeout, TimeUnit unit)设置最大等待时间。

java 中的线程池

线程池的实现原理

当提交一个新任务到线程池时，线程池的处理流程如下：

1. 线程池判断核心线程池里的线程是否都在执行任务。如果不是，则创建一个新的工作线程来执行任务。如果核心线程池里的线程都在执行任务，则进入下个流程。
2. 线程池判断工作队列是否已经满。如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。
3. 线程池判断线程池的线程是否都处于工作状态。如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务。

饱和策略

1. 什么时候会饱和？

当核心线程corePoolSize满且阻塞队列也满时才会判断当前线程数是否小于最大线程数，并决定是否创建新线程，如果创建的线程总数大于maximumPoolSize的时候，就会触发RejectedExecetionHandler。

2. 饱和策略有哪些？

   JDK主要提供了4种饱和策略供选择。4种策略都做为静态内部类在ThreadPoolExcutor中进行实现。

• AbortPolicy中止策略 丢弃任务并抛出RejectedExecutionException异常。
• DiscardPolicy抛弃策略 也是丢弃任务，但是不抛出异常
• DiscardOldestPolicy抛弃旧任务策略 丢弃队列最前面的任务，然后重新尝试执行任务(重复此过程)
• CallerRunsPolicy由调用线程处理该任务

合理的配置线程

1）配置线程池时CPU密集型任务可以少配置线程数，大概和机器的cpu核数相当，可以使得每个线程都在执行任务

2）IO密集型时，大部分线程都阻塞，需要多配置线程数，2cpu核数，非阻塞 IO 按需设置ncpu核数

3）有界队列和无界队列的配置需区分业务场景，建议配置有界队列。有界队列能增加系统的稳定性和预警能力。在一些可能会有爆发性增长的情况下使用无界队列。

4）任务非常多时，使用非阻塞队列使用CAS操作替代锁可以获得好的吞吐量。

线程池的监控

如果在线程中大量使用线程池，则有改要对线程池监控，方便在出现问题时，可以快速根据线程池的使用状态快速定位问题。

/**

* 线程池需要执行的任务数

*/

long taskCount = threadPoolExecutor.getTaskCount();

/**

* 线程池在运行过程中已完成的任务数

*/

long completedTaskCount = threadPoolExecutor.getCompletedTaskCount();

/**

* 曾经创建过的最大线程数

*/

long largestPoolSize = threadPoolExecutor.getLargestPoolSize();

/**

* 线程池里的线程数量

*/

long poolSize = threadPoolExecutor.getPoolSize();

/**

* 线程池里活跃的线程数量

*/
long activeCount = threadPoolExecutor.getActiveCount();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

java 并发实践

线上问题定位

问题定位

1: 首先使用TOP命令查看每个进程的情况，显示如下：

top - 22:27:25 up 463 days, 12:46, 1 user, load average: 11.80, 12.19, 11.79
 Tasks: 113 total, 5 running, 108 sleeping, 0 stopped, 0 zombie
 Cpu(s): 62.0%us, 2.8%sy, 0.0%ni, 34.3%id, 0.0%wa, 0.0%hi, 0.7%si, 0.2%st
 Mem: 7680000k total, 7665504k used, 14496k free, 97268k buffers
 Swap: 2096472k total, 14904k used, 2081568k free, 3033060k cached
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
 31177 admin 18 0 5351m 4.0g 49m S 301.4 54.0 935:02.08 java
 31738 admin 15 0 36432 12m 1052 S 8.7 0.2 11:21.05 nginx-proxy
1
2
3
4
5
6
7
8

我们的程序是Java应用，所以只需要关注COMMAND是Java的性能数据，COMMAND表示启动当前进程的命令，在Java进程这一行里可以看到CPU利用率是300％，不用担心，这个是当前机器所有核加在一起的CPU利用率。

2: 再使用Top的交互命令数字1查看每个CPU的性能数据。

top - 22:24:50 up 463 days, 12:43, 1 user, load average: 12.55, 12.27, 11.73
 Tasks: 110 total, 3 running, 107 sleeping, 0 stopped, 0 zombie
 Cpu0 : 72.4%us, 3.6%sy, 0.0%ni, 22.7%id, 0.0%wa, 0.0%hi, 0.7%si, 0.7%st
 Cpu1 : 58.7%us, 4.3%sy, 0.0%ni, 34.3%id, 0.0%wa, 0.0%hi, 2.3%si, 0.3%st
 Cpu2 : 53.3%us, 2.6%sy, 0.0%ni, 34.1%id, 0.0%wa, 0.0%hi, 9.6%si, 0.3%st
 Cpu3 : 52.7%us, 2.7%sy, 0.0%ni, 25.2%id, 0.0%wa, 0.0%hi, 19.5%si, 0.0%st
 Cpu4 : 59.5%us, 2.7%sy, 0.0%ni, 31.2%id, 0.0%wa, 0.0%hi, 6.6%si, 0.0%st
 Mem: 7680000k total, 7663152k used, 16848k free, 98068k buffers
 Swap: 2096472k total, 14904k used, 2081568k free, 3032636k cached
1
2
3
4
5
6
7
8
9

命令行显示了CPU4，说明这是一个5核的虚拟机，平均每个CPU利用率在60％以上。如果这里显示CPU利用率100％，则很有可能程序里写了一个死循环。这些参数的含义，可以对比下表：

3: 使用Top的交互命令H查看每个线程的性能信息。

PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
 31558 admin 15 0 5351m 4.0g 49m S 12.2 54.0 10:08.31 java
 31561 admin 15 0 5351m 4.0g 49m R 12.2 54.0 9:45.43 java
 31626 admin 15 0 5351m 4.0g 49m S 11.9 54.0 13:50.21 java
 31559 admin 15 0 5351m 4.0g 49m S 10.9 54.0 5:34.67 java
 31612 admin 15 0 5351m 4.0g 49m S 10.6 54.0 8:42.77 java
 31555 admin 15 0 5351m 4.0g 49m S 10.3 54.0 13:00.55 java
 31630 admin 15 0 5351m 4.0g 49m R 10.3 54.0 4:00.75 java
 31646 admin 15 0 5351m 4.0g 49m S 10.3 54.0 3:19.92 java
 31653 admin 15 0 5351m 4.0g 49m S 10.3 54.0 8:52.90 java
 31607 admin 15 0 5351m 4.0g 49m S 9.9 54.0 14:37.82 java
1
2
3
4
5
6
7
8
9
10
11

在这里可能会出现三种情况：

1. 第一种情况，某个线程一直CPU利用率100％，则说明是这个线程有可能有死循环，那么请记住这个PID。
2. 第二种情况，某个线程一直在TOP十的位置，这说明这个线程可能有性能问题。
3. 第三种情况，CPU利用率TOP几的线程在不停变化，说明并不是由某一个线程导致CPU偏高。

如果是第一种情况，也有可能是GC造成，我们可以用jstat命令看下GC情况,看看是不是因为持久代或年老代满了，产生Full GC，导致CPU利用率持续飙高，命令如下。

sudo /opt/java/bin/jstat -gcutil 31177 1000 5
 S0 S1 E O P YGC YGCT FGC FGCT GCT
 0.00 1.27 61.30 55.57 59.98 16040 143.775 30 77.692 221.467
 0.00 1.27 95.77 55.57 59.98 16040 143.775 30 77.692 221.467
 1.37 0.00 33.21 55.57 59.98 16041 143.781 30 77.692 221.474
 1.37 0.00 74.96 55.57 59.98 16041 143.781 30 77.692 221.474
 0.00 1.59 22.14 55.57 59.98 16042 143.789 30 77.692 221.481
1
2
3
4
5
6
7

这里解释一下 jstat 命令

# 查看11552进程内存统计情况
jstat -gc 11552

# 每隔1s打印一次，统计情况
jstat -gc 11552 1000

# 每隔1s打印一次，打印10次
jstat -gc 11552 1000 10 
1
2
3
4
5
6
7
8

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-r4aW7dmV-1661604160140)(D:\notebook\文档截图\jstat_gc命令.png)]

C：Capacity 

T：Time

U：Used

CCS：Campass Class Space

M：Metadata

O：Old

S：Survivor

E：Eden

Y：Young

S0C：第一个幸存区的大小

S1C：第二个幸存区的大小

S0U：第一个幸存区的使用大小

S1U：第二个幸存区的使用大小

EC：伊甸园区的大小

EU：伊甸园区的使用大小

OC：老年代大小

OU：老年代使用大小

MC：方法区大小

MU：方法区使用大小

CCSC:压缩类空间大小

CCSU:压缩类空间使用大小

YGC：年轻代垃圾回收次数

YGCT：年轻代垃圾回收消耗时间

FGC：老年代垃圾回收次数

FGCT：老年代垃圾回收消耗时间

GCT：垃圾回收消耗总时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

# 查看11552进程内存统计情况
jstat -gcutil 11552
 
# 每隔1s打印一次，统计情况
jstat -gcutil 11552 1000
 
# 每隔1s打印一次，打印10次
jstat -gcutil 11552 1000 10 
1
2
3
4
5
6
7
8

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-H9mEGxII-1661604160140)(D:\notebook\文档截图\jstat_gcutil命令.png)]

S0：幸存1区当前使用比例

S1：幸存2区当前使用比例

E：伊甸园区使用比例

O：老年代使用比例

M：元数据区使用比例

CCS：压缩使用比例

YGC：年轻代垃圾回收次数

FGC：老年代垃圾回收次数

FGCT：老年代垃圾回收消耗时间

GCT：垃圾回收消耗总时间
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

我们还可以把线程Dump下来，看看究竟是哪个线程，执行什么代码造成的CPU利用率高。执行以下命令，把线程dump到文件dump17里。

sudo -u admin /opt/java/bin/jstack  31177 > /home/tengfei.fangtf/dump17
1

jstack 命令说明
jstack pid > 输出的位置
1
2

dump出来内容的类似下面这段：

"http-0.0.0.0-7001-97" daemon prio=10 tid=0x000000004f6a8000 nid=0x555e in Object.wait() [0x0000000052423000]
   java.lang.Thread.State: WAITING (on object monitor)
        at java.lang.Object.wait(Native Method)
        - waiting on  (a org.apache.tomcat.util.net.AprEndpoint$Worker)
        at java.lang.Object.wait(Object.java:485)
        at org.apache.tomcat.util.net.AprEndpoint$Worker.await(AprEndpoint.java:1464)
        - locked  (a org.apache.tomcat.util.net.AprEndpoint$Worker)
        at org.apache.tomcat.util.net.AprEndpoint$Worker.run(AprEndpoint.java:1489)
        at java.lang.Thread.run(Thread.java:662)
1
2
3
4
5
6
7
8
9

dump出来的线程ID（nid）是十六进制的，而我们用TOP命令看到的线程ID是10进制的，所以我们要printf命令转换一下进制。然后用16进制的ID去dump里找到对应的线程。

printf "%x\n" 31558
 输出：7b46
1
2

Netstat 命令用于显示各种网络相关信息，如网络连接，路由表，接口状态 (Interface Statistics)，masquerade 连接，多播成员 (Multicast Memberships) 等等

常见参数

-a (all)显示所有选项，默认不显示LISTEN相关
-t (tcp)仅显示tcp相关选项
-u (udp)仅显示udp相关选项
-n 拒绝显示别名，能显示数字的全部转化成数字。
-l 仅列出有在 Listen (监听) 的服務状态

-p 显示建立相关链接的程序名
-r 显示路由信息，路由表
-e 显示扩展信息，例如uid等
-s 按各个协议进行统计
-c 每隔一个固定时间，执行该netstat命令。

如查看当前有多少数据库连接

$ netstat -nat | grep 3306 -c
12
表示已经连接了12个连接到数据库
1
2
3

参考文档

优化实战

1：查看下TCP连接状态，建立了800多个连接，需要尽量降低ESTABLISHED。

[tengfei.fangtf@ifeve ~]$ netstat -nat | awk '{print $6}' | sort | uniq -c | sort -n
 1 established)
 1 Foreign
 3 CLOSE_WAIT
 7 CLOSING
 14 FIN_WAIT2
 25 LISTEN
 39 LAST_ACK
 609 FIN_WAIT1
 882 ESTABLISHED
 10222 TIME_WAIT 
1
2
3
4
5
6
7
8
9
10
11

2：用jstack dump看看这些线程都在做什么。

sudo -u admin /opt/ifeve/java/bin/jstack 31177 > /home/tengfei.fangtf/dump17
1

3：统计下所有线程分别处于什么状态，发现大量线程处于WAITING(onobjectmonitor)状态

[tengfei.fangtf@ifeve ~]$ grep java.lang.Thread.State dump17 | awk '{print $2$3$4$5}' | sort | uniq -c
 39 RUNNABLE
 21 TIMED_WAITING(onobjectmonitor)
 6 TIMED_WAITING(parking)
 51 TIMED_WAITING(sleeping)
 305 WAITING(onobjectmonitor)
 3 WAITING(parking)
1
2
3
4
5
6
7

4：查看处于WAITING(onobjectmonitor)的线程信息，主要是jboss的工作线程在await。

"http-0.0.0.0-7001-97" daemon prio=10 tid=0x000000004f6a8000 nid=0x555e in Object.wait() [0x0000000052423000]
 java.lang.Thread.State: WAITING (on object monitor)
 at java.lang.Object.wait(Native Method)
 - waiting on <0x00000007969b2280> (a org.apache.tomcat.util.net.AprEndpoint$Worker)
 at java.lang.Object.wait(Object.java:485)
 at org.apache.tomcat.util.net.AprEndpoint$Worker.await(AprEndpoint.java:1464)
 - locked <0x00000007969b2280> (a org.apache.tomcat.util.net.AprEndpoint$Worker)
 at org.apache.tomcat.util.net.AprEndpoint$Worker.run(AprEndpoint.java:1489)
 at java.lang.Thread.run(Thread.java:662)
1
2
3
4
5
6
7
8
9

5：找到jboss的线程配置信息，将maxThreads降低到100

<maxThreads="250" maxHttpHeaderSize="8192"
 emptySessionPath="false" minSpareThreads="40" maxSpareThreads="75" maxPostSize="512000" protocol="HTTP/1.1"
 enableLookups="false" redirectPort="8443" acceptCount="200" bufferSize="16384"
 connectionTimeout="15000" disableUploadTimeout="false" useBodyEncodingForURI="true">
1
2
3
4

6：重启jboss，再dump线程信息，然后统计，WAITING(onobjectmonitor)的线程减少了170。

[tengfei.fangtf@ifeve ~]$ grep java.lang.Thread.State dump17 | awk '{print $2$3$4$5}' | sort | uniq -c
 44 RUNNABLE
 22 TIMED_WAITING(onobjectmonitor)
 9 TIMED_WAITING(parking)
 36 TIMED_WAITING(sleeping)
 130 WAITING(onobjectmonitor)
 1 WAITING(parking)
1
2
3
4
5
6
7