【程序员翻身计划】Java高性能编程第一章-Java多线程概述

2021-08-31 14:31:16

目标

重点：
- 线程安全的概念
- 线程通信的方式与应用
- reactor线程模型
- 线程数量的优化
- jdk常用命令
- Netty框架的作用
难点
- java运行的原理
- 同步关键字的原理
- AQS的抽象
- JUC的源码
- 网络编程的概念
- GC机制

class文件内容

文件开头有一个0xcafebabe特殊的标志。

包含版本、访问标志、常量池、当前类、超级类、接口、字段、方法、属性

【程序员翻身计划】Java高性能编程第一章-Java多线程概述

把class文件的信息存在方法区里面，有了类根据类创建对象，存储在堆内存中，垃圾回收就是这里。这是线程共享的部分，随虚拟机或者GC创建或销毁。除了这个区域还有线程独占空间，随线程生命周期而创建和销毁。

【程序员翻身计划】Java高性能编程第一章-Java多线程概述

方法区：用来存储加载的类信息、常量、静态变量、编译后的代码等数据。虚拟机规范中，这是一个逻辑区划，不同的虚拟机不同的实现。oracle的HotSpot在java7中，方法区放在永久代，java8放在元数据空间，并且通过GC机制对这个区域进行管理。
堆内存：分为老年代、新生代（Eden、From Survivor、To Survivor） JVM启动时创建，存放对象的实例。垃圾回收主要管理堆内存。

【程序员翻身计划】Java高性能编程第一章-Java多线程概述

虚拟机栈：每个线程都有一个私有的空间。线程栈由多个栈帧（Stack Frame）组成，一个线程会执行一个或多个方法，一个方法对应一个栈帧。

栈帧包括：局部变量表、操作数栈、动态链接、方法返回地址、附加信息。栈内存默认最大1M，超出抛出*Error
本地方法栈：使用Native本地方法准备的，超出也会报*Error，不同虚拟机厂商不同的实现。
程序计数器：记录当前线程执行字节码的位置，存储的是字节码指令地址，如果执行Native方法，计数器值会为空。

CPU同一时间只会执行一条线程中的指令。JVM多线程会轮流切换并分配CPU执行时间的方式。为了线程切换后，需要通过程序计数器来恢复正确的执行位置。

接下来是源文件编译后字节码相关的东西，暂不在本次笔记中记录。【记得有本书是字节码相关的解读，立个flag，日后学习！】

【程序员翻身计划】Java高性能编程第一章-Java多线程概述

线程状态

6个状态

new：尚未启动的线程的状态
runnable：可运行线程的线程状态，等待CPU调度
blocked：线程阻塞等待监视器锁定的状态，处于synchronized同步代码块或方法中被阻塞。
waiting：等待线程的状态，不带超时的方式：object.wait Thread.join LockSupport.pard
timed waiting : 具有指定等待时间的等待线程的线程状态。带超时的方式：Thread.sleep Object.wait Thread.join LockSupport.parkNanos LockSupport.parkUntil
Terminated：终止线程的状态，执行完毕或出现异常。

【程序员翻身计划】Java高性能编程第一章-Java多线程概述

案例1

//新建  运行  终止

System.out.println("#####第一种状态新建  运行  终止");

Thread thread1 = new Thread(new Runnable() {

    @Override

    public void run() {

        System.out.println("thread1当前状态："+Thread.currentThread().getState().toString());

        System.err.println("thread1执行了");

    }

});

System.out.println("没调用start方法，thread1当前状态："+thread1.getState().toString());

thread1.start();

Thread.sleep(2000);

System.out.println("等待两秒，thread1当前状态："+thread1.getState().toString());

复制代码

#####第一种状态新建  运行  终止

没调用start方法，thread1当前状态：NEW

thread1当前状态：RUNNABLE

thread1执行了

等待两秒，thread1当前状态：TERMINATED

复制代码

案例2

System.out.println("######第二种 新建 运行  等待 运行  终止（sleep）");

Thread thread2 = new Thread(new Runnable() {

    @Override

    public void run() {

        try {

            Thread.sleep(1500L);

        } catch (InterruptedException e) {

            // TODO Auto-generated catch block

            e.printStackTrace();

        }

        System.out.println("thread2当前状态："+Thread.currentThread().getState().toString());

        System.err.println("thread2执行了");

    }

});

Thread.sleep(2000);

System.out.println("没调用start方法，thread2当前状态：" + thread2.getState().toString());

thread2.start();

System.out.println("调用start方法，thread2当前状态：" + thread2.getState().toString());

Thread.sleep(200);

System.out.println("等待200毫秒，thread2当前状态：" + thread2.getState().toString());

Thread.sleep(3000);

System.out.println("等待3秒，thread2当前状态：" + thread2.getState().toString());

复制代码

######第二种 新建 运行  等待 运行  终止（sleep）

没调用start方法，thread2当前状态：NEW

调用start方法，thread2当前状态：RUNNABLE

等待200毫秒，thread2当前状态：TIMED_WAITING

thread2当前状态：RUNNABLE

thread2执行了

等待3秒，thread2当前状态：TERMINATED

复制代码

案例3

System.out.println("###第三种 新建  运行  阻塞  运行 终止");

Thread thread = new Thread(new Runnable() {

    @Override

    public void run() {

        synchronized (Test.class) {

            System.out.println("当前状态："+Thread.currentThread().getState().toString());

            System.out.println("执行了");

        }

    }

});

synchronized (Test.class) {

    System.out.println("没调用start方法，当前状态："+thread.getState().toString());

    thread.start();

    System.out.println("调用start方法，当前状态："+thread.getState().toString());

    Thread.sleep(200);

    System.out.println("200毫秒后，当前状态："+thread.getState().toString());

}

Thread.sleep(3000);

System.out.println("3秒后，当前状态："+thread.getState().toString());

复制代码

###第三种 新建  运行  阻塞  运行 终止

没调用start方法，当前状态：NEW

调用start方法，当前状态：RUNNABLE

200毫秒后，当前状态：BLOCKED

当前状态：RUNNABLE

执行了

3秒后，当前状态：TERMINATED

复制代码

线程终止

stop()

线程不安全，会强行终止线程的所有锁定。
interrupt()

如果目标线程在调用Object class的wait join sleep方法时被阻塞，那么interrupt会生效，该线程的中断状态将被清除，抛出interruptedException异常。

如果目标线程是被IO或者NIO中的channel阻塞，IO操作会被中断或者返回特殊异常值。达到终止的目的。

如果以上条件都不满足，则会设置此线程的中断状态。

通过状态位来判断

public class StopThread extends Thread{

	public volatile static boolean flag = true;

	public static void main(String[] args) throws InterruptedException {

		new Thread(()-> {

			while(flag) {

				try {

					System.out.println("运行中");

					Thread.sleep(10000);

				} catch (InterruptedException e) {

					// TODO Auto-generated catch block

					e.printStackTrace();

				}

			}

		}).start();

		Thread.sleep(3000);

		flag = false;

		System.out.println("结束");

	}

}

复制代码

CPU缓存及内存屏障

CPU有三级缓存，从123到内存再到硬盘。但是存在一个问题，如果多核cpu读取同样的数据进行缓存计算，最终写入主内存的是以哪个为准？

这个时候就出来了一个缓存一致性协议，单个cpu对缓存中的数据做了改动，需要通知给其他cpu。

CPU还有一个性能优化手段，运行时指令重排，把读缓存命令优先执行。

两个问题：

缓存中的数据与主内存中的数据并不是实时同步的，各个cpu间缓存的数据也不是实时同步的，在同一个时间点，各个cpu看到的同一内存地址的数据的值可能是不一致的。
多核多线程中，指令逻辑无法分辨因果关系，可能出现乱序执行。

解决办法：内存屏障

写内存屏障：在指令后插入Store Barrier，能让写入缓存中的最新数据更新写入主内存，让其他线程可见。
读内存屏障：在指令前插入Load Barrier，可以让高速缓存中的数据失效，强制重新从主内存中加载数据。

线程通信

要想实现多个线程之间的协同，如线程执行先后顺序，获取某个线程执行的结果等，设计线程之间相互通信。

文件共享
网络共享
共享变量
jdk提供的线程协调API

~~suspend/resume~~、wait/notify、park/unpark

JDK中对于需要多线程协作的，提供了对应API支持，典型场景是：生产者-消费者模型（线程阻塞、线程唤醒）

suspend/resume

同步代码中使用，suspend挂起之后并不会释放锁，容易出现死锁。
suspend比resume后执行

被弃用。

wait/notify notifyAll

只能由同一对象锁的持有者线程调用，也就是写在同步块里，否则会抛出illegalMonitorStateException异常。

wait：加入该对象的等待集合中，并且放弃当前持有的对象锁。

虽然wait会自动解锁，但是对顺序有要求，如果在notify被调用之后才开始wait方法的调用，线程会永远处于WAITING状态。

//正常的wait

public void waitNotify() throws Exception {

    new Thread(()->{

        if(baozidian == null) {

            synchronized (this) {

                System.out.println("进入等待");

            }

        }

        System.out.println("买到包子");

    }).start();

    Thread.sleep(3000);

    baozidian = new Object();

    synchronized (this) {

        this.notify();

        System.out.println("通知");

    }

}

结果：

进入等待

买到包子

通知

复制代码

park/unpark

线程调用park则等待许可，unpark为指定线程提供许可。

不要求方法的调用顺序。但不会释放锁，所以在同步代码块中使用可能会死锁。

/** 死锁的park/unpark */

public void parkUnparkDeadLockTest() throws Exception {

    // 启动线程

    Thread consumerThread = new Thread(() -> {

        if (baozidian == null) { // 如果没包子，则进入等待

            System.out.println("1、进入等待");

            // 当前线程拿到锁，然后挂起

            synchronized (this) {

                LockSupport.park();

            }

        }

        System.out.println("2、买到包子，回家");

    });

    consumerThread.start();

    // 3秒之后，生产一个包子

    Thread.sleep(3000L);

    baozidian = new Object();

    // 争取到锁以后，再恢复consumerThread

    synchronized (this) {

        LockSupport.unpark(consumerThread);

    }

    System.out.println("3、通知消费者");

}

结果：

1、进入等待

复制代码

注意：最好不要使用if语句来判断是否进入等待状态。

官方建议应该在循环体中检查等待状态，原因是处于等待状态的线程可能会收到错误警报和伪唤醒。

伪唤醒是指线程因为更底层的原因导致的。

线程封闭

并不是所有时候都要用到共享数据，shuju被封闭在各自的线程中，就不需要同步。

具体体现有：ThreadLocal、局部变量

ThreadLocal

是一个线程级别的变量，每个线程都有一个ThreadLocal，就是每个线程都拥有了自己独立的一个变量，竞争条件被彻底消除了，在并发模式下，是绝对安全的变量。

线程池

线程在java中是一个对象，更是操作系统的资源，创建销毁都需要时间。
java对象占用堆内存，操作系统线程占用系统内存，根据jvm规范，一个线程默认最大栈大小是1M，线程过多会消耗很多内存。
操作系统需要频繁切换线程上下文。

----->线程池就是为了解决这些问题。

线程池概念

线程池管理器：创建并管理，创建、销毁线程池、添加新任务
工作线程：在没有任务时处于等待状态，可以循环执行任务
任务接口：每个任务必须实现的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等。
任务队列：存放没有处理的任务。提供一种缓冲机制。

【程序员翻身计划】Java高性能编程第一章-Java多线程概述

线程池API-接口定义和实现类

类型	名称	描述
接口	Executor	最上层的接口，定义了`执行任务的方法execute`
接口	ExecutorService	继承了Executor接口，拓展了Callable、Future、关闭方法
接口	ScheduledExecutorService	继承了ExecutorService接口，增加了定时任务相关的方法
实现类	ThreadPoolExecutor	`基础、标准的线程池实现`
实现类	ScheduledThreadPoolExecutor	继承了ThreadPoolExecutor，实现了 ScheduledExecutorService中相关定时任务的方法

代码示例

公共代码块：

/**

     * 测试：提交15个执行时间需要三秒，看线程池的情况

     * @param threadPoolExecutor

     */

public void testCommon(ThreadPoolExecutor threadPoolExecutor) throws Exception{

    for (int i=0;i<15;i++){

        int n = i;

        threadPoolExecutor.submit(new Runnable() {

            @Override

            public void run() {

                try {

                    System.out.println("开始执行："+n);

                    Thread.sleep(3000l);

                    System.err.println("执行结束："+n);

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

            }

        });

        System.out.println("提交任务成功:"+i);

    }

    Thread.sleep(500l);

    System.out.println("当前线程池的数量："+threadPoolExecutor.getPoolSize());

    System.out.println("当前等待队列的数量："+threadPoolExecutor.getQueue().size());

    Thread.sleep(15000l);

    System.out.println("当前线程池的数量："+threadPoolExecutor.getPoolSize());

    System.out.println("当前等待队列的数量："+threadPoolExecutor.getQueue().size());

}

复制代码

测试方法1：

/**

 * 1、线程池信息： 核心线程数量5，最大数量10，*队列，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的

 *

 * @throws Exception

 */

public void threadPoolExecutorTest1() throws Exception{

    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5,10,5, TimeUnit.SECONDS,

            new LinkedBlockingQueue<Runnable>());

    testCommon(threadPoolExecutor);

}

//预计的结果：线程池数量5，其他进入等待队列

复制代码

测试方法1输出结果：

提交任务成功:0

开始执行：0

提交任务成功:1

开始执行：1

提交任务成功:2

开始执行：2

提交任务成功:3

提交任务成功:4

提交任务成功:5

提交任务成功:6

提交任务成功:7

提交任务成功:8

提交任务成功:9

提交任务成功:10

提交任务成功:11

提交任务成功:12

提交任务成功:13

提交任务成功:14

开始执行：3

开始执行：4

当前线程池的数量：5

当前等待队列的数量：10

执行结束：2

执行结束：0

执行结束：4

执行结束：1

执行结束：3

开始执行：5

开始执行：6

开始执行：7

开始执行：8

开始执行：9

开始执行：10

开始执行：11

开始执行：12

开始执行：13

开始执行：14

执行结束：5

执行结束：6

执行结束：8

执行结束：7

执行结束：9

执行结束：13

执行结束：10

执行结束：14

执行结束：12

执行结束：11

当前线程池的数量：5

当前等待队列的数量：0

复制代码

这里有一个问题就是，最大线程数量设置的是10，当前线程池的数量为什么达不到最大线程数量？

这就需要对execute的过程有个了解。

【程序员翻身计划】Java高性能编程第一章-Java多线程概述

测试方法2：

/**

     * 2、 线程池信息： 核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的

     *

     * @throws Exception

     */

    public void threadPoolExecutorTest2() throws Exception{

        // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。

        // 如果不指定拒绝策略，默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,

                new LinkedBlockingQueue<Runnable>(3), new RejectedExecutionHandler() {

            @Override

            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {

                System.out.println("任务决绝执行。");

            }

        });

        testCommon(threadPoolExecutor);

    }

//执行预期结果：

//线程池数量5,3个进入等待，这时候核心线程数量和队列都满了，会加开5个任务线程（注意，5秒后没任务执行会销毁），因为最大线程是10

//最大10+等待队列3   总共13，剩下两个拒绝执行

复制代码

测试方法3：Executors.newFixedThreadPool(int nThreads)

对于*队列，最大线程数量实际上是不起作用的。

    /**

     * 3、 线程池信息： 核心线程数量5，最大数量5，*队列，超出核心线程数量的线程存活时间：5秒

     *

     * @throws Exception

     */

    private void threadPoolExecutorTest3() throws Exception {

        // 和Executors.newFixedThreadPool(int nThreads)一样的

        ThreadPoolExecutor threadPoolExecutor1 = (ThreadPoolExecutor) Executors.newFixedThreadPool(5);

//        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,

//                new LinkedBlockingQueue<Runnable>());

        testCommon(threadPoolExecutor1);

        // 预计结：线程池线程数量为：5，超出数量的任务，其他的进入队列中等待被执行

    }

复制代码

Executors.newFixedThreadPool()的内部实现实际上就是注释掉的部分：

    public static ExecutorService newFixedThreadPool(int nThreads) {

        return new ThreadPoolExecutor(nThreads, nThreads,

                                      0L, TimeUnit.MILLISECONDS,

                                      new LinkedBlockingQueue<Runnable>());

    }

复制代码

测试方法4：Executors.newCachedThreadPool()

此种方法适用于不可预估数量的情况

/**

     * 4、 线程池信息：

     * 核心线程数量0，最大数量Integer.MAX_VALUE，SynchronousQueue队列，超出核心线程数量的线程存活时间：60秒

     *

     * @throws Exception

     */

    private void threadPoolExecutorTest4() throws Exception {

        // SynchronousQueue，实际上它不是一个真正的队列，因为它不会为队列中元素维护存储空间。与其他队列不同的是，它维护一组线程，这些线程在等待着把元素加入或移出队列。

        // 在使用SynchronousQueue作为工作队列的前提下，客户端代码向线程池提交任务时，

        // 而线程池中又没有空闲的线程能够从SynchronousQueue队列实例中取一个任务，

        // 那么相应的offer方法调用就会失败（即任务没有被存入工作队列）。

        // 此时，ThreadPoolExecutor会新建一个新的工作者线程用于对这个入队列失败的任务进行处理（假设此时线程池的大小还未达到其最大线程池大小maximumPoolSize）。

        // 和Executors.newCachedThreadPool()一样的

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,

                new SynchronousQueue<Runnable>());

        testCommon(threadPoolExecutor);

        // 预计结果：

        // 1、 线程池线程数量为：15，超出数量的任务，其他的进入队列中等待被执行

        // 2、 所有任务执行结束，60秒后，如果无任务可执行，所有线程全部被销毁，池的大小恢复为0

        Thread.sleep(60000L);

        System.out.println("60秒后，再看线程池中的数量：" + threadPoolExecutor.getPoolSize());

    }

复制代码

 测试方法4输出结果：

提交任务成功:0

提交任务成功:1

提交任务成功:2

提交任务成功:3

提交任务成功:4

提交任务成功:5

提交任务成功:6

提交任务成功:7

提交任务成功:8

提交任务成功:9

提交任务成功:10

提交任务成功:11

提交任务成功:12

提交任务成功:13

提交任务成功:14

开始执行：3

开始执行：2

开始执行：6

开始执行：7

开始执行：10

开始执行：11

开始执行：0

开始执行：1

开始执行：5

开始执行：4

开始执行：8

开始执行：9

开始执行：12

开始执行：13

开始执行：14

当前线程池的数量：15

当前等待队列的数量：0

执行结束：3

执行结束：2

执行结束：6

执行结束：7

执行结束：10

执行结束：9

执行结束：0

执行结束：1

执行结束：5

执行结束：4

执行结束：14

执行结束：8

执行结束：11

执行结束：12

执行结束：13

当前线程池的数量：15

当前等待队列的数量：0

60秒后，再看线程池中的数量：0

复制代码

测试方法5：一次性定时任务

/**

     * 5、 定时执行线程池信息：3秒后执行，一次性任务，到点就执行 <br/>

     * 核心线程数量5，最大数量Integer.MAX_VALUE，DelayedWorkQueue延时队列，超出核心线程数量的线程存活时间：0秒

     *

     * @throws Exception

     */

public void threadPoolExecutorTest5() throws Exception{

    //Executors.newScheduledThreadPool() 一样的

    ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(5);

    scheduledThreadPoolExecutor.schedule(new Runnable() {

        @Override

        public void run() {

            System.out.println("任务被执行，现在时间："+ DateUtil.now());

        }

    },3000,TimeUnit.MILLISECONDS);

    System.out.println("定时任务，提交成功，时间是："+DateUtil.now());

}

复制代码

测试方法5输出结果：

定时任务，提交成功，时间是：2021-11-27 13:42:43

任务被执行，现在时间：2021-11-27 13:42:46

复制代码

测试方法6：周期定时任务

/**

     * 6、 定时执行线程池信息：线程固定数量5 ，<br/>

     * 核心线程数量5，最大数量Integer.MAX_VALUE，DelayedWorkQueue延时队列，超出核心线程数量的线程存活时间：0秒

     *

     * @throws Exception

     */

public void threadPoolExecutorTest6() throws Exception{

    ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(5);

    //第一种方式：scheduleAtFixedRate，如果执行时间超过了周期时间

    //执行完毕后，立即执行，不考虑延迟时间

    scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable() {

        @Override

        public void run() {

            try {

                Thread.sleep(3000l);

            }catch (InterruptedException e){

                e.printStackTrace();

            }

            System.out.println("任务1被执行，现在时间："+DateUtil.now());

        }

    },2000,1000,TimeUnit.MILLISECONDS);

    //第二种方式，scheduleWithFixedDelay

    //如果执行时间超过了周期时间，执行完毕后，加上延迟时间后再执行

    scheduledThreadPoolExecutor.scheduleWithFixedDelay(new Runnable() {

        @Override

        public void run() {

            try {

                Thread.sleep(3000l);

            }catch (InterruptedException e){

                e.printStackTrace();

            }

            System.out.println("任务2被执行，现在时间："+DateUtil.now());

        }

    },2000,1000,TimeUnit.MILLISECONDS);

}

复制代码

测试方法6输出结果：

//可以看出，任务1每隔3秒执行一次，任务2每隔4秒执行一次

任务1被执行，现在时间：2021-11-27 14:08:45

任务2被执行，现在时间：2021-11-27 14:08:45

任务1被执行，现在时间：2021-11-27 14:08:48

任务2被执行，现在时间：2021-11-27 14:08:49

任务1被执行，现在时间：2021-11-27 14:08:51

任务2被执行，现在时间：2021-11-27 14:08:53

任务1被执行，现在时间：2021-11-27 14:08:54

任务1被执行，现在时间：2021-11-27 14:08:57

任务2被执行，现在时间：2021-11-27 14:08:57

复制代码

终止线程的两种方式

scheduledThreadPoolExecutor.shutdown();

//第二种会返回尚未执行的任务

List<Runnable> runnableList = scheduledThreadPoolExecutor.shutdownNow();

复制代码

本文同步公众号【刘墨泽】，欢迎大家关注聊天！

码农公寓

目标

class文件内容

线程状态

6个状态

案例1

案例2

案例3

线程终止

CPU缓存及内存屏障

线程通信

suspend/resume

wait/notify notifyAll

park/unpark

线程封闭

ThreadLocal

线程池

线程池概念

线程池API-接口定义和实现类

代码示例

测试方法1：

测试方法2：

测试方法3：Executors.newFixedThreadPool(int nThreads)

测试方法4：Executors.newCachedThreadPool()

测试方法5：一次性定时任务

测试方法6：周期定时任务

终止线程的两种方式

相关文章