qidao123.com技术社区-IT企服评测·应用市场

标题: 百万架构师第四十五课：并发编程的基础｜JavaGuide [打印本页]
作者: 泉缘泉    时间: 2025-3-1 22:41
标题: 百万架构师第四十五课：并发编程的基础｜JavaGuide
课程目标

1. 多线程的发展汗青
2. 线程的应用
3. 并发编程的基础
4. 线程安全的问题
特定的指令，计算机不会存储指令，把指令写下来，一次性读取指令，批处理。
然后我们需要把批处理举行隔离、生存它的进度。
进程 —> 线程
单核CPU 只有可能会有一个进程去执行。
什么情况下应该利用多线程

线程出现的目的是什么？解决进程中多使命的及时性的问题？其实简单来说，就是解决“阻塞”的问题。阻塞的意思就是程序运行到某个函数或过程后等待某些事件发生而临时制止 CPU 占用的情况，也就是说会使得 CPU 闲置。另有一些场景就是比如对于一个函数中的运算逻辑的性能问题，我们可以 通过多线程的技术，使得一个函数中的多个逻辑运算通过多线程技术达到一个并行执行，从而提高性能。
CPU 架构图解：


所以，多线程终极解决的就是“等待”的问题，所以简单总结的利用场景
Tomcat 7 以前的 I/O 模型

多线程的应用场景
tomcat 模型：


多个客户端都是阻塞的，我只有处理完一个请求才能接收下一个请求。然后客户端就会阻塞。所以 Tomcat 采用了多线程的技术。利用了多线程的技术实现了非阻塞。

如何应用多线程

在 JAVA 中有多个方式来实现多线程。继续 Thread 类、实现 Runable 接口、利用 ExecutorService 、Callable、Future 实现带返回结果的多线程。
继续 Thread 类创建线程

​        Thread 类本质上是实现了 Runable 接口的一个实例，代表一个线程的实例。启动线程的唯一方法就是通过 Thread 类的 start() 方法。 start() 方法是一个 native 方法。它会启动一个新线程，并执行 run() 方法。这种实现多线程很简单，通过本身的类直接 extends Thread ， 并重写 run() 方法，就可以启动新线程并执行本身定义的 run() 方法。
  1. public class MyThread extends Thread{
  2.     public static void main(String[] args) {
  3.         new MyThread().start();
  4.         new MyThread().start();
  5.     }
  6.     @Override
  7.     public void run() {
  8.         System.out.println("MyThrea run().....");
  9.     }
  10. }
复制代码
实现 Runable 接口创建线程

​        如果本身的类已经继续了另一个类，就无法直接继续 Thread，此时，可以实现 Runable 接口。
  1. public class RunableDemo implements Runnable {
  2.     public static void main(String[] args) {
  3.         new Thread(new RunableDemo()).start();
  4.         new Thread(new RunableDemo()).start();
  5.     }
  6.     @Override
  7.     public void run() {
  8.         System.out.println("[" + Thread.currentThread().getName() + "]" + "runable My run().....");
  9.     }
  10. }
复制代码
  1. [Thread-0]runable My run().....
  2. [Thread-1]runable My run().....
复制代码
错误的写法：
  1. public class RunableDemo implements Runnable {
  2. //    public static void main(String[] args) {
  3. //        new Thread(new RunableDemo()).start();
  4. //        new Thread(new RunableDemo()).start();
  5. //    }
  7.     public static void main(String[] args) {
  8.         new RunableDemo().run();
  9.         new RunableDemo().run();
  10.     }
  11.     @Override
  12.     public void run() {
  13.         System.out.println("[" + Thread.currentThread().getName() + "]" + "runable My run().....");
  14.     }
  15. }
复制代码
  1. [main]runable My run().....
  2. [main]runable My run().....
复制代码
实现 Callable 接口通过 FutureTask 包装器来创建 Thread 线程

​        有的时候，我们可能需要让异步执行的线程在执行完以后，提供一个返回值到当前的主线程，主线程需要这个值举行后续的逻辑处理，那么这个时候，就需要带返回值的线程了。
  1. /***
  2. * 当你想要异步的线程执行你的某一个逻辑，那么在这个运行结束以后
  3. * 我想要拿到子线程运行的结果
  4. */
  5. public class CallableDemo implements Callable<String> {
  6.     public static void main(String[] args) throws Exception {
  7.         ExecutorService executorService = Executors.newSingleThreadExecutor();
  9.         CallableDemo callableDemo = new CallableDemo();
  11.         Future<String> future = executorService.submit(callableDemo);
  12.         /***
  13.          * 这里可以写其他的业务
  14.          * 去写其他东西
  15.          */
  16.         String returnValue = future.get(); // 这个地方在阻塞
  17.         System.out.println(returnValue);
  18.         executorService.shutdown();
  19.     }
  21.     @Override
  22.     public String call() throws Exception {
  23.         return "darain" + 1;
  24.     }
  25. }
复制代码
如何把多线程用得优雅

合理地利用异步操作，可以大大地提拔程序的处理性能，下面这个案例，如何看过 zookeeper 源码的同学应该看到过。
通过阻塞队列以及多线程的方式，实现对请求的异步化处理，提拔处理性能。
模仿多个线程处理同一个请求
  1. @Data
  2. public class Request {
  3.     private String name;
  4. }
复制代码
  1. public interface RequestProcessor {
  2.     void processorRequest(Request requset);
  3. }
复制代码
  1. @RequiredArgsConstructor
  2. public class PrintProcessor extends Thread implements RequestProcessor {
  3.     LinkedBlockingQueue<Request> linkedBlockingQueue = new LinkedBlockingQueue<>();
  4.     private final RequestProcessor nextProcess;
  6.     @Override
  7.     public void processorRequest(Request requset) {
  8.         linkedBlockingQueue.add(requset);
  9.     }
  11.     @Override
  12.     public void run() {
  13.         while (true) {
  14.             try {
  15.                 Request requset = linkedBlockingQueue.take();
  16.                 out.println("[" + Thread.currentThread().getName() + "] " + "print Data:" + requset);
  17.                 nextProcess.processorRequest(requset);
  18.             } catch (InterruptedException e) {
  19.                 e.printStackTrace();
  20.             }
  21.         }
  22.     }
  23. }
复制代码
  1. @RequiredArgsConstructor
  2. public class SaveProcessor extends Thread implements RequestProcessor {
  4.     LinkedBlockingQueue<Request> linkedBlockingQueue = new LinkedBlockingQueue<>();
  6.     @Override
  7.     public void processorRequest(Request requset) {
  8.         linkedBlockingQueue.add(requset);
  9.     }
  11.     @Override
  12.     public void run() {
  13.         while (true) {
  14.             try {
  15.                 Request requset = linkedBlockingQueue.take();
  16.                 System.out.println("[" + Thread.currentThread().getName() + "] " + "save data:" + requset);
  17.             } catch (InterruptedException e) {
  18.                 e.printStackTrace();
  19.             }
  20.         }
  21.     }
  22. }
复制代码
  1. /***
  2. * 我们去处理的时候，用异步线程去处理。
  3. * 当我们把一个请求丢过来的时候，不是直接去处理，而是通过异步线程去处理。
  4. * zookeeper 就是类似的处理，一方面，你可以通过你的处理把职责划分开。
  5. * 一方面你可以通过异步线程的处理去提升你程序的性能
  6. * 合理地利用你 CPU 的资源
  7. *
  8. * 这个和 zookeeper 里边非常像
  9. */
  10. public class Demo {
  11.     private final PrintProcessor printProcessor;
  13.     public Demo() {
  14.         SaveProcessor saveProcessor = new SaveProcessor();
  15.         saveProcessor.start();
  16.         printProcessor = new PrintProcessor(saveProcessor);
  17.         printProcessor.start();
  18.     }
  20.     public static void main(String[] args) {
  21.         Request requset = new Request();
  22.         requset.setName("darian");
  23.         new Demo().doTest(requset);
  24.     }
  26.     public void doTest(Request request) {
  27.         printProcessor.processorRequest(request);
  28.     }
  29. }
复制代码

就像一个链表一样地，上一个对象的引用指向下一个对象。是不会乱序的。
线程的基础知识

​        线程作为操作系统调度的最小单元，并且能够让多线程同时执行，极大地提高了程序的性能，在多核的环境下的优势更加显着。但是在多线程的利用过程中如果对它的特性和原理不够了解的话，就容易造成各种问题。
线程的状态（六种）

JAVA 线程既然能够创建，那么也会销毁，所以线程是存在生命周期的。那么我们接下来从线程的生命周期开始去了解线程。
线程一共六种状态
（NEW、RUNNABLE、BLOCKED、WAITING、TIME_WAITING、TERMINATED）
NEW

​        初始状态，线程被构建，但是还没有调用 #start 方法
RUNNABLE

​        运行状态，JAVA 线程把操作系统中的就绪和运行两种状态同一称为 "运行中"
BLOCKED

阻塞，表示线程进如等待状态，也就是线程因为某种原因放弃了 CPU 的利用权，阻塞也分为几种情况。
等待   （waiting） 是我们的线程调用了一个 #wait 方法，实际上也会变成一个阻塞。就是我们没有办法继续去运行线程了。
TIME_WAITING

​         超时等待状态，超时以后自动返回
TERMINATED

​        制止状态，表示当前线程执行完毕
线程运行状态图：


线程的运行状态有两种状态，
不存在就绪的状态，只是为了描述它的一个状态。
打开 Thread 类，搜索 state 有哪些状态，它写得很清楚。
当运行中的线程的时间片被 CPU 抢占的时候，那么它又会变成一个就绪状态。
线程执行完就是制止。
synchroninzed 就是让这个线程获得锁。获得锁，就意味着，其他线程在调用这个方法的时候，它会阻塞。当我们获得锁的时候。比如说我们现在有两个线程。第一个 T1 线程访问同步代码块。同步代码块里面，首先它会获得一个锁。当 T2 线程进来以后，它是没有办法获得锁的。
线程状态：
  1. public class ThreadStatusDemo {
  2.     public static void main(String[] args) {
  3.         new Thread(() -> {
  4.             while (true) {
  5.                 try {
  6.                     TimeUnit.SECONDS.sleep(100);
  7.                 } catch (InterruptedException e) {
  8.                     e.printStackTrace();
  9.                 }
  10.             }
  11.         }, "timewaiting").start();
  13.         new Thread(() -> {
  14.             while (true) {   // 我们在一个循环里边获得一个锁
  15.                 synchronized (ThreadStatusDemo.class) {
  16.                     try {
  17.                         // 然后调用 wait()  方法，是因为它调用 wait 方法之前必须要获得锁
  18.                         ThreadStatusDemo.class.wait();
  19.                     } catch (InterruptedException e) {
  20.                         e.printStackTrace();
  21.                     }
  22.                 }
  23.             }
  24.         }, "waiting").start();
  27.         new Thread(new blockDemo(), "blockDemo-0").start();
  28.         new Thread(new blockDemo(), "blockDemo-1").start();
  29.     }
  31.     static class blockDemo extends Thread {
  32.         @Override
  33.         public void run() {
  34.             synchronized (blockDemo.class) {
  35.                 while (true) {
  36.                     try { // 100 秒，一直让它阻塞
  37.                         TimeUnit.SECONDS.sleep(100);
  38.                     } catch (InterruptedException e) {
  39.                         e.printStackTrace();
  40.                     }
  41.                 }
  42.             }
  43.         }
  44.     }
  45. }
复制代码
通过相应的命令显示线程状态：

我们在写线程的时候，最好定义一个名称。我们去查看问题的时候，有利于我们去排查问题。
阻塞状态，blocked 当 synchronized 加锁的情况下，两个线程同时去访问一个方法，这个时候，就会存在 阻塞。
JPS 是 JDK 1.5 以后，显示所有 JAVA 进程的命令。
jstack 30112  可以查看线程的状态。
  1. "DestroyJavaVM" #18 prio=5 os_prio=0 tid=0x0000000002a02800 nid=0x697c waiting on condition [0x0000000000000000]
  2.    java.lang.Thread.State: RUNNABLE
  4. "blockDemo-1" #17 prio=5 os_prio=0 tid=0x0000000029066800 nid=0x5f74 waiting for monitor entry [0x0000000029bff000]
  5.    java.lang.Thread.State: BLOCKED (on object monitor)
  7. "blockDemo-0" #15 prio=5 os_prio=0 tid=0x0000000029065800 nid=0x36f8 waiting on condition [0x0000000029aff000]
  8.    java.lang.Thread.State: TIMED_WAITING (sleeping)
  10. "waiting" #13 prio=5 os_prio=0 tid=0x0000000029061000 nid=0x7bd0 in Object.wait() [0x00000000299fe000]
  11.    java.lang.Thread.State: WAITING (on object monitor)
  13. "timewaiting" #12 prio=5 os_prio=0 tid=0x0000000029094800 nid=0x6bd4 waiting on condition [0x00000000298fe000]
  14.    java.lang.Thread.State: TIMED_WAITING (sleeping)
  16. "Service Thread" #11 daemon prio=9 os_prio=0 tid=0x0000000027540800 nid=0x8310 runnable [0x0000000000000000]
  17.    java.lang.Thread.State: RUNNABLE
  19. "C1 CompilerThread3" #10 daemon prio=9 os_prio=2 tid=0x000000002747e000 nid=0x5344 waiting on condition [0x0000000000000000]
  20.    java.lang.Thread.State: RUNNABLE
  22. "C2 CompilerThread2" #9 daemon prio=9 os_prio=2 tid=0x000000002747d000 nid=0x24c0 waiting on condition [0x0000000000000000]
  23.    java.lang.Thread.State: RUNNABLE
  25. "C2 CompilerThread1" #8 daemon prio=9 os_prio=2 tid=0x0000000027475800 nid=0x7c30 waiting on condition [0x0000000000000000]
  26.    java.lang.Thread.State: RUNNABLE
  28. "C2 CompilerThread0" #7 daemon prio=9 os_prio=2 tid=0x0000000027474800 nid=0x5c78 waiting on condition [0x0000000000000000]
  29.    java.lang.Thread.State: RUNNABLE
  31. "Monitor Ctrl-Break" #6 daemon prio=5 os_prio=0 tid=0x000000002745c800 nid=0xde0 runnable [0x00000000289fe000]
  32.    java.lang.Thread.State: RUNNABLE
  34. "Attach Listener" #5 daemon prio=5 os_prio=2 tid=0x00000000273ba000 nid=0x3434 waiting on condition [0x0000000000000000]
  35.    java.lang.Thread.State: RUNNABLE
  37. "Signal Dispatcher" #4 daemon prio=9 os_prio=2 tid=0x0000000027411800 nid=0x839c runnable [0x0000000000000000]
  38.    java.lang.Thread.State: RUNNABLE
  40. "Finalizer" #3 daemon prio=8 os_prio=1 tid=0x00000000273a3800 nid=0x79bc in Object.wait() [0x00000000286fe000]
  41.    java.lang.Thread.State: WAITING (on object monitor)
  43. "Reference Handler" #2 daemon prio=10 os_prio=2 tid=0x0000000002afa000 nid=0x4e7c in Object.wait() [0x00000000285ff000]
  44.    java.lang.Thread.State: WAITING (on object monitor)
复制代码
我们很多时候，要多发现线程的信息。
线程的启动和制止

你怎么去启动一个线程？制止？
#start  native 方法，告诉 JVM 去启动一个线程。然后调用 #run 方法去执行。
#stop 方法是不建议利用的。 @Deprecated ！！它就像我们在 Linux 系统中，kill 命令一样，就是我不知道我当前这个线程是不是还在运行，有没有还没处理完的。没有处理完的话，我强制关闭，就会出现一些数据问题，和一些不可预测地问题出现。 #susped， #resume 。
怎么样优雅的关闭？我们关闭 Tomcat 也好，关闭一些进程也好，我们都会提供一些优雅的方式去关闭。一些指令去执行，一般的中间件都会做一个操作，一般都会先去制止后续的请求进来，然后等待正在运行的线程执行完以后优雅地制止掉。
#interrupt 优雅中断的方式。

​        当其他线程通过调用当前线程的 #interrupt 方法，表示向当前线程打个招呼，告诉他可以中断线程的执行了，至于什么时候中断，取决于当前线程本身。线程通过检查自身是或否被中断来举行响应，可以通过 isIntrrupted() 来判断是否被中断。
实现线程制止的逻辑：
  1. public class InterruptDemo {
  2.     private static int i;
  3.     public static void main(String[] args) throws InterruptedException {
  4.         Thread thread = new Thread(() -> {
  5.             // 我去判断是否中断这个线程
  6.             while (!Thread.currentThread().isInterrupted()) {
  7.                 i++;
  8.             }
  9.             System.out.println(i);
  10.         }, "interruptDemo");
  11.         thread.start();
  12.         TimeUnit.SECONDS.sleep(1);
  13.         // 通过线程的 interrupt  设置标识为 true
  14.         System.out.println(thread.isInterrupted());
  15.         thread.interrupt();
  16.         System.out.println(thread.isInterrupted());
  17.     }
  18. }
复制代码
这种通过表示为或者中断操作的方式能够使线程在制止时有机会去清理资源，而不是武断地将线程制止。因此更加安全和优雅。
Thread.interrupted

通过 interrupt，设置了一个标识告诉线程可以制止运行了。线程中还提供了静态方法 Thread.interrupted() 对设置中断标识的线程复位。比如在线程，外边的线程调用 thread.interrupt 来设置中断标识，而在线程里边，又通过 Thread.interrupted 把线程的标识举行了复位。
  1. public static void interrupt1() throws InterruptedException {
  2.     Thread thread = new Thread(() -> {
  3.         while (true) {
  4.             boolean interrupted = Thread.currentThread().isInterrupted();
  5.             if (interrupted) {
  6.                 System.out.println("before:" + interrupted);
  7.                 Thread.interrupted(); // 对线程进行复位，中断标识为 false
  8.                 System.out.println("after:" + Thread.currentThread().isInterrupted());
  9.             }
  10.         }
  11.     });
  12.     thread.start();
  13.     TimeUnit.SECONDS.sleep(1);
  14.     thread.interrupt(); // 设置中断标识为 true
  15. }
复制代码
  1. before:true
  2. after:false
复制代码
其他的线程复位

​        除了通过 Thread.interrupted 方法对线程中断标识举行复位以外，另有一种被动复位的场景，就是对抛出 interruptedException 非常的方法，在 interruptedException 抛出之前， JVM 会先把线程的中断标识位清除，然后会抛出 InterruptedException 这个时候，如果调用 #isInterrupted 方法，将会返回 false。
  1. public static void interrupt2() throws InterruptedException {
  2.     Thread thread = new Thread(() -> {
  3.         while (true) {
  4.             try {
  5.                 Thread.sleep(10000);
  6.             } catch (InterruptedException e) {
  7.                 // 抛出该异常，会将复位表示设置为 false
  8.                 e.printStackTrace();
  9.             }
  10.         }
  11.     });
  12.     thread.start();
  13.     thread.interrupt(); // 将复位表示设置为 true
  14.     TimeUnit.SECONDS.sleep(1);
  15.     System.out.println("before:" + thread.isInterrupted());
  16.     TimeUnit.SECONDS.sleep(1);
  17.     System.out.println("after:" + thread.isInterrupted());
  18. }
复制代码

通过指令的方式，volatile boolean isStop = false; 这样的一个方式，也是可以的。通过内存的可见。
interrupt 和我们设置标志变量的方式是一样的。
java.lang.Thread#interrupt
thread.cpp
  1. bool Thread::is_interrupted(Thread* thread, bool clear_interrupted) {
  2.   debug_only(check_for_dangling_thread_pointer(thread);)
  3.   // Note:  If clear_interrupted==false, this simply fetches and
  4.   // returns the value of the field osthread()->interrupted().
  5.   return os::is_interrupted(thread, clear_interrupted);
  6. }
复制代码
os_linux.cpp
  1. void os::interrupt(Thread* thread) {
  2.     assert(Thread::current() == thread || Threads_lock->owned_by_self(), "possibility of dangling Thread pointer");
  3.     OSThread* osthread = thread->osthread();
  4.     if (!osthread->interrupted()) {
  5.         osthread->set_interrupted(true);
  6.         // More than one thread can get here with the same value of osthread,
  7.         // resulting in multiple notifications. We do, however, want the store
  8.         // to interrupted() to be visible to other threads before we execute unpark().
  9.         OrderAccess::fence();
  10.         ParkEvent * const slp = thread->_SleepEvent ;
  11.         if (slp != NULL) slp->unpark() ;
  12.     }
  13.     // For JSR166. Unpark even if interrupt status already was set
  14.     if (thread->is_Java_thread())
  15.         ((JavaThread*)thread)->parker()->unpark();
  16.     ParkEvent * ev = thread->_ParkEvent ;
  17.     if (ev != NULL) ev->unpark() ;
  18. }
复制代码
内存屏蔽 fence() ，让标志位改变，让所有线程看见，和 volatile 一个意思。
unpark() 线程。
其实就是通过 unpark 去唤醒
Thread#interrupted  是一个静态方法，对设置的中断标识的线程举行复位。
线程的制止方法之 2

​        除了通过 #interrupt 标识去中断线程以外，我们可以通过 ：
​        定义一个 volatile 修饰的成员变量，来控制线程的制止。这实际上是应用了 volatile 实现多线程之间的共享变量可见性这一特点来实现的。
  1. public class ThreadStopDemo3 {
  2.     // 这种和 interrupted 方式是一样的。
  3.     private static volatile boolean stop = true;
  5.     public static void main(String[] args) throws InterruptedException {
  6.         Thread thread = new Thread(() -> {
  7.             int i = 0;
  8.             while (!stop) {
  9.                 i++;
  10.             }
  11.         });
  12.         thread.start();
  13.         System.out.println("begin start thread");
  14.         Thread.sleep(1000);
  15.         stop = true;
  16.     }
  17. }
复制代码
线程的安全性

熟悉这三个问题。
可见性
  1. /***
  2. * 可见性问题
  3. */
  4. public class VisableDemo {
  5.     // 加上 volatile 之后，才可以停止。
  6.     private volatile static boolean stop = false;
  8.     public static void main(String[] args) throws InterruptedException {
  9.         Thread thread = new Thread(() -> {
  10.             int i = 0;
  11.             while (!stop) {
  12.                 i++;
  13.             }
  14.         });
  15.         thread.start();
  16.         TimeUnit.SECONDS.sleep(1);
  17.         stop = true;
  18.     }
  19. }
复制代码
原子性

[code]/*** * */public class AutomicDemo {    private static int count = 0;    public static void main(String[] args) throws InterruptedException {        for (int i = 0; i < 1000; i++) {            new Thread(AutomicDemo::inc).start();        }        Thread.sleep(4000);        System.out.println("y运行结果：" + count);        // y运行结果：952    (



欢迎光临 qidao123.com技术社区-IT企服评测·应用市场 (https://dis.qidao123.com/) Powered by Discuz! X3.4