并发编程（一）：初识多线程及线程通信

缺乏、安全感 2023-10-10 10:07 169阅读 0赞

一，线程三种启动方式

  1. 1，Thread方式
  3. package com.gupao.concurrent;
  5. /**
  6.  * @author pj_zhang
  7.  * @create 2019-09-24 21:06
  8.  **/
  9. public class StartThread extends Thread {
  11.     @Override
  12.     public void run() {
  13.         System.out.println("Thread方式执行线程。。。");
  14.     }
  16.     public static void main(String[] args) {
  17.         // 普通方式执行
  18.         new StartThread().start();
  19.         // lamdo表达式执行
  20.         new Thread(() -> {
  21.             System.out.println("lamdo: Thread方式执行线程。。。");
  22.         }).start();
  23.     }
  24. }
  26. 2，Runnable方式
  28. package com.gupao.concurrent;
  30. /**
  31.  * @author pj_zhang
  32.  * @create 2019-09-24 21:12
  33.  **/
  34. public class StartRunnable implements Runnable {
  35.     @Override
  36.     public void run() {
  37.         System.out.println("Runnable方式执行线程。。。");
  38.     }
  40.     public static void main(String[] args) {
  41.         new Thread(new StartRunnable()).start();
  42.     }
  43. }
  45. 3，Callable方式 -- 带返回值
  47.     \* 不同于Thread和Runnable方式，Callable执行线程后可以返回线程执行结果
  49.     \* 通过返回结果Funture去get()结果集时，会阻塞直到结果返回
  51. package com.gupao.concurrent;
  53. import java.util.concurrent.*;
  55. /**
  56.  * @author pj_zhang
  57.  * @create 2019-09-24 21:13
  58.  **/
  59. public class StartCallable implements Callable {
  60.     @Override
  61.     public Object call() throws Exception {
  62.         TimeUnit.SECONDS.sleep(3);
  63.         return "Callable方式执行线程";
  64.     }
  66.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  67.         ExecutorService service = new ThreadPoolExecutor(
  68.                 20, 20, 0L,
  69.                 TimeUnit.SECONDS, new LinkedBlockingQueue<>());
  70.         Long startTime;
  71.         Long endTime;
  72.         System.out.println("开始执行时间：" + (startTime = System.currentTimeMillis()));
  73.         Future<?> future = service.submit(new StartCallable());
  74.         String result = (String) future.get();
  75.         System.out.println("执行结束时间" +  (endTime = System.currentTimeMillis()));
  76.         System.out.println(result + ", cast: " + (endTime - startTime));
  77.     }
  78. }

二，线程声明周期（六种线程状态）

  1. 1，线程状态
  3.     \* NEW（线程初始化）
  5.     \* RUNNABLE（线程运行）
  7.         a，Ready（就绪）
  9.         b，Running（运行）
  11.         c，NEW -(start())-> RUNNABLE
  13.         d，Ready -线程调度-> Running
  15.         e，Running -(yield()/线程调度)-> Ready
  17.     \* TERMINATED（线程销毁）
  19.     \* BLOKED（线程阻塞状态，锁抢占）
  21.          a，synchorized加锁
  23.     \* WAITING（线程等待）
  25.          a，wait()
  27.          b，join()
  29.          c，LockSupport.park()
  31.     \* TIME\_WAITING（线程限时等待）
  33.          a，wait(Long)
  35.          b，sleep(Long)
  37. 2，线程状态转换演示\_代码
  39. package com.gupao.concurrent;
  41. import java.util.concurrent.locks.Lock;
  42. import java.util.concurrent.locks.ReentrantLock;
  44. /**
  45.  * @author pj_zhang
  46.  * @create 2019-09-24 21:42
  47.  **/
  48. public class ThreadStatusTest {
  50.     public static void main(String[] args) {
  51.         // 运行时线程
  52.         new Thread(() -> {
  53.             for (;;) {}
  54.         }, "THREAD_RUNNING").start();
  55.         // 等待线程
  56.         new Thread(() -> {
  57.             try {
  58.                 Object object = new Object();
  59.                 synchronized (object) {
  60.                     object.wait();
  61.                 }
  62.             } catch (InterruptedException e) {
  63.                 e.printStackTrace();
  64.             }
  65.         }, "THREAD_WAITING").start();
  66.         // 定时等待线程
  67.         new Thread(() -> {
  68.             try {
  69.                 Object object = new Object();
  70.                 synchronized (object) {
  71.                     object.wait(60 * 60 * 1000);
  72.                 }
  73.             } catch (InterruptedException e) {
  74.                 e.printStackTrace();
  75.             }
  76.         }, "TIME_WAITING_THREAD").start();
  77.         Lock lock = new ReentrantLock();
  78.         // 未阻塞线程
  79.         new Thread(() -> {
  80.             synchronized (lock) {
  81.                 for (; ; ) {}
  82.             }
  83.         }, "NOT_BLOCKING_THREAD").start();
  84.         // 阻塞线程
  85.         new Thread(() -> {
  86.             try {
  87.                 Thread.sleep(1000);
  88.                 synchronized (lock) {
  89.                     for (;;) {}
  90.                 }
  91.             } catch (InterruptedException e) {
  92.                 e.printStackTrace();
  93.             }
  94.         }, "BLOCKING_THREAD").start();
  95.     }
  96. }
  98. 3，线程状态转换演示\_状态查看
  100.     \* 通过 jps 命令查看正在执行的线程PID

2019092422281056.png

  1.     \* 通过 jstack pid 命令，查看正在执行的所有线程及其状态

20190924222836870.png

  1.     \* 从中可以依次看到定义中的几种线程状态

20190924222914515.png

20190924222925414.png

20190924222943513.png

20190924223000419.png

2019092422301841.png

三，线程中断

  1. 1，线程中断：thread.interrupt()
  3. 2，判断线程是否中断：Thread.currentThread().isInterrupted()
  5. 3，线程中断代码演示；如下代码，interrupt()提供了一种可控的方式对线程进行中断，执行中断操作后，被中断线程可以根据当前业务运行情况在对应安全点对线程中断操作进行响应
  7. package com.gupao.concurrent;
  9. /**
  10.  * @author pj_zhang
  11.  * @create 2019-09-24 22:33
  12.  **/
  13. public class ThreadInterruptTest {
  15.     private static int i = 0;
  17.     public static void main(String[] args) throws InterruptedException {
  18.         Thread thread = new Thread(() -> {
  19.                 // 线程初始化后，默认中断状态为false
  20.                 // 等其他线程对当前线程执行中断操作后，线程状态变为true
  21.                 for (; !Thread.currentThread().isInterrupted(); ) {
  22.                     i++;
  23.                 }
  24.                 System.out.println(i);
  25.                 // 打印当前线程中断状态
  26.                 System.out.println(Thread.currentThread().isInterrupted());
  27.             });
  28.         thread.start();
  29.         // 主线程睡眠一秒后，对线程进行中断操作
  30.         Thread.sleep(1000);
  31.         thread.interrupt();
  32.     }
  33. }

20190924224052189.png

  1. 4，线程等待中的线程中断演示，会抛出 InterruptedException 异常，(异常后会对线程中断状态进行复位，在下一部分解释)
  3. package com.gupao.concurrent;
  5. /**
  6.  * @author pj_zhang
  7.  * @create 2019-09-24 22:33
  8.  **/
  9. public class ThreadWaitingInterruptTest {
  11.     private static int i = 0;
  13.     public static void main(String[] args) throws InterruptedException {
  14.         Thread thread = new Thread(() -> {
  15.             try {
  16.                 // 线程等待十秒，主线程1秒后中断，则必定触发中断异常
  17.                 Thread.sleep(10000);
  18.             } catch (InterruptedException e) {
  19.                 // 异常后， 线程会自动复位
  20.                 // 主线程调用interrupt后，线程中断状态从false -> true
  21.                 // 由于触发异常，则对线程中断重新赋值为初始状态false
  22.                 System.out.println("响应线程中断。。。，线程中断状态："
  23.                         + Thread.currentThread().isInterrupted());
  24.             }
  25.         });
  26.         thread.start();
  27.         // 主线程睡眠一秒后，对线程进行中断操作
  28.         Thread.sleep(1000);
  29.         thread.interrupt();
  30.     }
  31. }

20190924224708652.png

四，线程复位

  1. 1，线程中断是给执行一个中断信号，告诉执行线程已经进行了线程中断操作，具体后续操作由执行线程进行后续处理，执行线程在接收到线程中断信号后，可以对线程中断状态进行复位；
  3. 2，线程复位两种方式
  5.     a，API复位：Thread.interrupted()
  7. package com.gupao.concurrent;
  9. /**
  10.  * @author pj_zhang
  11.  * @create 2019-09-24 22:33
  12.  **/
  13. public class ThreadInterruptTest {
  15.     private static int i = 0;
  17.     public static void main(String[] args) throws InterruptedException {
  18.         Thread thread = new Thread(() -> {
  19.                 // 线程初始化后，默认中断状态为false
  20.                 // 等其他线程对当前线程执行中断操作后，线程状态变为true
  21.                 for (; !Thread.currentThread().isInterrupted(); ) {
  22.                     i++;
  23.                 }
  24.                 System.out.println(i);
  25.                 // 线程复位前中断状态
  26.                 System.out.println("before: " + Thread.currentThread().isInterrupted());
  27.                 // 线程复位
  28.                 Thread.interrupted();
  29.                 // 线程复位后中断状态
  30.                 System.out.println("after：" + Thread.currentThread().isInterrupted());
  31.             });
  32.         thread.start();
  33.         // 主线程睡眠一秒后，对线程进行中断操作
  34.         Thread.sleep(1000);
  35.         thread.interrupt();
  36.     }
  37. }

20190924225212598.png

  1.     b，异常复位：等待中的线程接收到异常信号后，会抛出 InterruptedException 异常，捕获异常后线程状态已经由 JVM 复位，可继续进行后续处理，异常复位已经在上一部分演示；

五，线程通信

  1. 1，线程等待和线程唤醒：wait()/notify()
  3. package com.gupao.concurrent;
  5. /**
  6.  * @author pj_zhang
  7.  * @create 2019-09-24 23:04
  8.  **/
  9. public class ThreadTelTest extends Thread {
  11.     // 定义初始数量
  12.     private static int i = 100;
  14.     // 定义初始标志位
  15.     // 此处模拟两道线程运行，通过简单标志位实现
  16.     // 如果确实需要通过某种方式实现线程顺序执行，可以通过分布式锁实现
  17.     // 参考之前博文中的zookeeper和redis分布式锁
  18.     private static boolean flag = false;
  20.     // 定义同步锁对象
  21.     private static Object lock = new Object();
  23.     public static void main(String[] args) {
  24.         // 两道线程同时执行
  25.         new Thread(new ThreadTelTest(), "售票员A").start();
  26.         new Thread(new ThreadTelTest(), "售票员B").start();
  27.     }
  29.     @Override
  30.     public void run() {
  31.         try {
  32.             for (;;) {
  33.                 synchronized (lock) {
  34.                     if (i > 0) {
  35.                         // flag为true，说明已经存在线程执行过，
  36.                         // 当前线程为新线程或者被唤醒线程，且执行过的线程目前处于等待中
  37.                         if (flag) {
  38.                             System.out.println(Thread.currentThread().getName() + "卖出第" + i-- + "张票");
  39.                             // 线程执行完成后，唤醒处于等待中线程
  40.                             lock.notify();
  41.                             // 当前线程等待，并释放锁对象，被唤醒的线程会直接抢占锁对象
  42.                             lock.wait();
  43.                         } else {
  44.                             // flag为false，说明当前线程第一次进入，执行完业务代码后，修改flag表示为true
  45.                             System.out.println(Thread.currentThread().getName() + "卖出第" + i-- + "张票");
  46.                             flag = true;
  47.                             // 线程等待，实现两道线程交替执行，因为是第一个进入的线程，不需要唤醒线程
  48.                             lock.wait();
  49.                         }
  50.                     } else {
  51.                         System.out.println("卖完了");
  52.                         break;
  53.                     }
  54.                 }
  55.             }
  56.         } catch (InterruptedException e) {
  57.             e.printStackTrace();
  58.         }
  59.     }
  60. }
  62. 2，线程排队：join()
  64.     \* 线程排队执行代码演示
  66. package com.gupao.concurrent;
  68. /**
  69.  * @author pj_zhang
  70.  * @create 2019-09-24 22:52
  71.  **/
  72. public class ThreadJoinTest {
  74.     public static void main(String[] args) throws InterruptedException {
  75.         Thread t1 = new Thread(() -> {
  76.             try {
  77.                 // 通过不规则定时，演示线程随机执行
  78.                 Thread.sleep(200);
  79.                 System.out.println("线程一执行结束。。。");
  80.             } catch (InterruptedException e) {
  81.                 e.printStackTrace();
  82.             }
  83.         });
  84.         Thread t2 = new Thread(() -> {
  85.             try {
  86.                 Thread.sleep(20);
  87.                 System.out.println("线程二执行结束。。。");
  88.             } catch (InterruptedException e) {
  89.                 e.printStackTrace();
  90.             }
  91.         });
  92.         Thread t3 = new Thread(() -> {
  93.             System.out.println("线程三执行结束。。。");
  94.         });
  95.         // 通过 join 方法，保证线程排队执行
  96.         t1.start();
  97.         t1.join();
  98.         t2.start();
  99.         t2.join();
  100.         t3.start();
  101.         t3.join();
  102.     }
  103. }
  105.     \* join()方法内部；
  107. public final synchronized void join(long millis)
  108.     throws InterruptedException {
  109.         long base = System.currentTimeMillis();
  110.         long now = 0;
  112.         if (millis < 0) {
  113.             throw new IllegalArgumentException("timeout value is negative");
  114.         }
  116.         if (millis == 0) {
  117.             // 从这一步可以看到，如果当前线程依旧存活，则阻塞主线程
  118.             // 通过自旋一直等待直到线程执行结束
  119.             while (isAlive()) {
  120.                 wait(0);
  121.             }
  122.         } else {
  123.             // 这一部分表示限时阻塞
  124.             while (isAlive()) {
  125.                 long delay = millis - now;
  126.                 if (delay <= 0) {
  127.                     break;
  128.                 }
  129.                 wait(delay);
  130.                 now = System.currentTimeMillis() - base;
  131.             }
  132.         }
  133.     }
  135.     \* 从方法源码看，join()通过对主线程的阻塞实现线程方法的顺序执行，如果在上一步直接start()三道线程，再对线程逐一进行join()，则线程三会瞬间抢占时间片并执行，从而乱序，如图

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE5NzYzODg_size_16_color_FFFFFF_t_70

发表评论

表情：
评论列表 （有 0 条评论，169人围观）

还没有评论，来说两句吧...

相关阅读

    相关 并发：线程初识

     一个程序同时执行多个任务。 通常，每一个任务称为一个线程 （ thread ) , 它是线程控制的简称。 可以同时运行一个以上线程的程序称为多线程程序 （ multithre

    野性酷女野性酷女/ 0 赞/ 314 阅读