ShymeanV0.8.0

初识Java多线程编程

发表于| 分类于编程语言

在前端开发工作中，基本上没有接触到多线程编程，最近在整理Android开发，感觉需要深入地学习一下多线程的知识，因此整理了一下Java中的多线程编程知识点。

<!--more-->

一个在运行的应用就是一个进程Process，一个进程可以包含一个或多个线程Thread，操作系统的最小调度单位是线程。

多进程比多线程更加稳定，一个进程的崩溃不会影响其他进程，但一个线程的崩溃会导致其整个进程的崩溃多线程的应用场景。创建进程比创建线程的开销大，进程之间的通信也比线程通信慢

从Java程序的角度来看：一个Java程序实际上是一个JVM进程，其主线程用来运行main()方法，在main方法内部可以启动多个线程。

参考

1. 创建线程

下面是几种创建线程的方法

1.1. Thread

可以通过继承Thread并重写run()方法来创建一个线程

// 继承Thread
class MyThread extends Thread {
    @Override
    public void run() {
        System.out.println("start new thread!");
    }
}

public class Main {
    public static void main(String[] args) {
        System.out.println("before main");
          // Thread t = new MyThread(); 
          // 下面这种方式更简洁
        Thread thread = new Thread(() -> {
            System.out.println("start thread!");
        });

        thread.start(); // 启动新线程
        System.out.println("after main");
    }
}

初学时需要理解的概念：线程的运行时机由操作系统决定，应用程序本身无法确定线程的调用顺序。这跟常规的程序顺序流差别很大！

也就是说，上面的代码中，无法确定after main和start thread!这两个谁会被先输出。

可以对线程设定优先级thread.setPriority，但也无法保证高优先级的线程肯定会先被执行。

在后续章节会进一步探讨线程的运行顺序、共享数据等行为。

1.2. Runnable

除了直接继承Thread，也可以通过实现Runnable接口来创建线程

class MyRunnable implements Runnable {
    @Override
    public void run() {
        System.out.println("run in thread ");
    }
}

public class Main {
    public static void main(String[] args) throws InterruptedException {
        System.out.println("before");

        MyRunnable r1 = new MyRunnable();
        Thread t = new Thread(r1);
        t.start();
          // t.join()
        System.out.println("after");
    }
}

实际上Thread就是一个实现了Runnable的类。

1.3. Callable

上面两种创建线程的方式，存在一个缺点：在执行完任务之后无法直接获取执行结果，只有通过共享变量或者使用线程通信的方式来传递执行结果，比较麻烦。

因此Java提供了Callable和Future，通过他们可以获取线程执行完成之后的结果

import java.util.concurrent.*;

public class Main {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        Task task = new Task();
        Future<Integer> result = executor.submit(task);
        executor.shutdown();


        try {
              // get方法会产生阻塞直到任务执行完毕为止
            System.out.println("task result:" + result.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        System.out.println("done");
    }
}

class Task implements Callable<Integer> {
    @Override
    public Integer call() throws Exception {
        System.out.println("task call");
        Thread.sleep(10);
        int sum = 0;
        for (int i = 0; i < 100; i++)
            sum += i;
        return sum;
    }
}

2. 结束线程

Thread.State定义了线程状态

  • NEW，new Thread()新创建未开始的线程
  • RUNNABLE，调用start()之后运行中的线程
  • BLOCKED，阻塞状态，等待锁的释放
  • WAITING，等待状态，调用Object的wait()、Thread的join()、LockSupport的park()方法之后，线程会等待其他线程处理特殊的行为，直到调用notifyAll()之后
  • TIMED_WAITING，有时间的等待状态，调用Thread的sleep()等方法后
  • TERMINATED，终止状态

等线程的run方法运行结束之后，线程就结束了。

在某些场景下，线程会持续做某些工作，所以经常会看见run()方法里面是一个循环

public class Main {
    public static void main(String[] args) {
        new Thread(() -> {
            while (true) {
                try {
                    Thread.sleep(1000);
                    System.out.println("loop...");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

这种线程需要等待其他线程通知结束，结束一个线程有多种方式

2.1. interrupt

第一种方式是调用interrupt()方法，在线程内可以通过isInterrupted()判断当前线程是否被中断。

public class Main {
    public static void main(String[] args) throws InterruptedException {
        Thread t = new MyThread();
        t.start();
        Thread.sleep(10); // 暂停
        t.interrupt(); // 中断t线程
        t.join(); // 等待t线程结束
        System.out.println("end");
    }
}

class MyThread extends Thread {
    public void run() {
        int n = 0;
        while (!this.isInterrupted()) {
            System.out.println("n is: " + ++n);
        }
    }
}

InterruptedException是一个检测异常，在大部分时候会打印或者忽略这个错误。

2.2. 线程标志位

另外一种方式是设置一个标志位，在线程的run方法中判断该标志位是否需要退出

public class Main {
    public static void main(String[] args) throws InterruptedException {
        MyThread t = new MyThread();
        t.start();
        Thread.sleep(10); // 暂停
        t.running = false; // 手动设置标志位为false
        t.join();
        System.out.println("end");
    }
}

class MyThread extends Thread {
    public volatile boolean running = true;
    public void run() {
        int n = 0;
        while (running) {
            System.out.println("n is: " + ++n);
        }
    }
}

这里需要注意volatile关键字，这个关键字会告诉虚拟机，

  • 每次访问该变量时，会获取主内存中最新值
  • 每次修改变量后，将值回写到主内存

这样可以保存每个线程访问到的变量值是统一的。这样，当一个线程修改了某个共享变量的值，其他线程能够立刻看到修改后的值。

3. 线程同步

多线程的特点在于：线程之间经常需要读取和操作共享数据，和导致代码的编写和调试更加复杂。

3.1. 概念

由于操作系统调用的不确定性，如果多个线程需要读写某个公共变量，就可能导致出现不一样的结果。

考虑下面代码，在main()方法结束之后，Data.count的结果会是什么呢？

class Data {
    public static int count = 1;
}

class AddThread extends Thread {
    public volatile boolean running = true;

    public void run() {
        Data.count += 2;
    }
}

class MultiThread extends Thread {
    public volatile boolean running = true;

    public void run() {
        Data.count *= 10;
    }
}

public class Main {
    public static void main(String[] args) throws InterruptedException {
        AddThread addThread = new AddThread();
        MultiThread multiThread = new MultiThread();
        addThread.start();
        multiThread.start();

        System.out.println("count is:" + Data.count);
    }
}

反复运行上面的代码，最后的输出可能得到很多种结果

  • 1，主线程先执行
  • 3，先运行了addThread.run再运行了主线程
  • 10，先运行了multiThread.run再运行了主线程
  • 12，先运行multiThread.run，再运行addThread.run，最后运行主线程
  • 30，先运行addThread.run，再运行multiThread.run，最后运行主线程

为了保证得到确定的结果，比如我们想得到30，可以使用join方法手动排列每个线程的运行顺序

public class Main {
    public static void main(String[] args) throws InterruptedException {
        AddThread addThread = new AddThread();
        MultiThread multiThread = new MultiThread();
        addThread.start();
        addThread.join();

        multiThread.start();
        multiThread.join();
        System.out.println("count is:" + Data.count);
    }
}

但手动排序让我们舍弃了多线程的优点，因此我们需要其他的方案来保证线程同步。

这种运行的不确定就是多线程编程中最复杂的部分。因此在多线程编程下，需要通过某种机制保证程序按照我们预期的结果执行。这就是

线程同步： 协同步调，让线程按照预定的先后次序运行。

注意这里不要把同步理解成同时进行，而应该是“协同”、“互相配合”。

比如线程1和线程2用来合作完成某个任务，按照职责分离，线程1和线程2可以各自负责某部分工作。在线程1执行到某个时刻，需要依赖线程2的运行的结果，然后停下来等待线程2运行完毕，再将结果传递给线程1继续运行，最后完成某个任务。

接下来看看实现线程同步的几种方式。

3.2. synchronized锁机制

先看下面这个问题

public class Main {
    public static void main(String[] args) throws Exception {
        var addThread = new AddThread();
        var descThread = new DescThread();
        addThread.start();
        descThread.start();
        addThread.join();
        descThread.join();
        System.out.println(Data.count);
    }
}

class Data {
    public static int count = 0;
}

class AddThread extends Thread {
    public void run() {
        for (int i = 0; i< 10000; i++) { 
          Data.count += 1; 
        }
    }
}

class DescThread extends Thread {
    public void run() {
        for (int i = 0; i < 10000; i++) { 
          Data.count -= 1; 
        }
    }
}

反复运行，基本上每次都可以得到不同的结果（可以将循环次数修改为10000甚至更大，用来得到更随机的结果）。

这是什么原因呢？

首先需要明确：线程run方法里面的代码并不是从头到尾一次性运行完毕的，这也是为什么写一个死循环看起来并不会像单线程那样让程序一直挂起。

在操作系统的调度下，addThread.run运行到某个时候，可能就跑去运行decThread.run，接着某个时候又回到addThread.run，周而复始，直到两个线程都运行完毕。

对于共享变量Data.count而言

Data.count += 1
等价于
Data.count = Data.count + 1

这行表达式有三个步骤指令：

  • 读取Data.count
  • 计算Data.count + 1
  • 将计算结果赋值给Data.count

对于Data.count -= 1同理。

操作系统有可能在这三个步骤中的任何一个步骤切换到其他线程出去，比如可能addThread中已经计算到了Data.count + 1的结果为50，这个时候跑去运行在decThread，下次切换回来后重新赋值，相当于这段时间内decThread运算的结果就被覆盖了。

因此对于某些包含多个运算步骤的代码块，我们期望操作系统一次性让他们运行完再进行切换。

这时候需要使用synchronized关键字，这个关键字接收一个对象，对于同一个对象而言，保证代码块在任意时刻只有一个线程能执行，这个传入的对象参数也被称为对象锁。

class Data {
    public static final Object lock = new Object();
    public static int count = 0;
}

class AddThread extends Thread {
    public void run() {
        for (int i = 0; i < 10000; i++) {
            synchronized (Data.lock) {
                Data.count += 1;
            }
        }
    }
}

class DecThread extends Thread {
    public void run() {
        for (int i = 0; i < 10000; i++) {
            synchronized (Data.lock) {
                Data.count -= 1;
            }
        }
    }
}

这样不管运行多少次，最后得到的都是同一个结果：0，也就是预期的结果。

synchronized约束的代码块无法执行并发性，同时加锁和解锁也会消耗时间，因此会代码性能下降。

JVM规范定义了几种为原子操作的单条代码，原子性代码无需使用synchronized

  • 基本类型（long和double除外）赋值，例如：int a = b
  • 引用类型赋值，例如：List<int> list = list2

包含多条代码的代码块不具备原子性，如果包含共享变量的操作，则需要进行synchronized同步

在线程中显式地使用synchronized看起来不太优雅，也容易犯错，需要将其封装起来

public class Main {
    public static void main(String[] args) throws Exception {
          var instance = Data.getInstance();

        var addThread = new AddThread();
        var descThread = new DescThread();

        addThread.start();
        descThread.start();
        addThread.join();
        descThread.join();
        System.out.println(instance.count);
    }
}

class Data {
    public static Data instance;
    public int count = 0;

    public static Data getInstance() {
        if (Data.instance == null) {
            Data.instance = new Data();
        }
        return Data.instance;
    }

    public void add() {
        synchronized (this) {
            count += 1;
        }
    }

    public void desc() {
        synchronized (this) {
            count -= 1;
        }
    }
}

class AddThread extends Thread {
    public void run() {
        for (int i = 0; i < 10000; i++) {
            Data.getInstance().add();
        }
    }
}

class DescThread extends Thread {
    public void run() {
        for (int i = 0; i < 10000; i++) {
            Data.getInstance().desc();
        }
    }
}

synchronized除了修饰代码块，也可以直接修饰整个方法

class Data {
    public static Data instance;
    public int count = 0;

    public synchronized static Data getInstance() {
        if (Data.instance == null) {
            Data.instance = new Data();
        }
        return Data.instance;
    }

      // 与synchronized (this) 一致
    public synchronized void add() {
        count += 1;
    }

    public synchronized void desc() {
        count -= 1;
    }
}

封装完成之后，在线程里面使用的调用方就无需关心共享变量的同步问题了。

如果一个类被设计为允许多线程正确访问，这个类就是“线程安全”的（thread-safe）。

也许是出于性能考虑，Java中的大部分类都是非线程安全的，需要开发者手动处理。

3.2.1. 死锁

对于同一个对象锁而言，同一时间只能被一个线程持有，只有当线程将锁释放之后，才能被其他线程持有。

当存在多个锁的时候，如果线程无法正确获取和释放锁，就会导致死锁。

下面是一种可能出现死锁的情况

public class Main {
    private static Object lockA = new Object();
    private static Object lockB = new Object();

    public static void main(String[] args) {
        new Thread(() -> {
            synchronized (lockA) {
                System.out.println(Thread.currentThread() + "get a");
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + "waiting get b");
                synchronized (lockB) {
                    System.out.println(Thread.currentThread() + "get b");
                }
            }
        }, "线程 1").start();
        new Thread(() -> {
            synchronized (lockB) {
                System.out.println(Thread.currentThread() + "get b");
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + "waiting get a");
                synchronized (lockA) {
                    System.out.println(Thread.currentThread() + "get a");
                }
            }
        }, "线程 2").start();
    }
}

运行代码后就可以发现，程序在输出下面内容之后就卡住了

Thread[线程 2,5,main]get b
Thread[线程 1,5,main]get a
Thread[线程 1,5,main]waiting get b
Thread[线程 2,5,main]waiting get a

分析一下代码

  • 线程1进入，获得lockA
  • 线程2进入，获取lockB

然后

  • 线程1尝试获得lockB，但lockB还未被释放，死锁
  • 线程2尝试获得lockA，但lockA还未被释放，死锁

查了一下资料，下面是形成死锁的4个必要条件

  • 互斥条件：线程(进程)对于所分配到的资源具有排它性，即一个资源只能被一个线程(进程)占用，直到被该线程(进程)释放
  • 请求与保持条件：一个线程(进程)因请求被占用资源而发生阻塞时，对已获得的资源保持不放。
  • 不剥夺条件：线程(进程)已获得的资源在末使用完之前不能被其他线程强行剥夺，只有自己使用完毕后才释放资源。
  • 循环等待条件：当发生死锁时，所等待的线程(进程)必定会形成一个环路（类似于死循环），造成永久阻塞

发生死锁后，只有重启JVM才能解除。我们需要谨慎编写代码，破坏上述4个条件中的其中一个，就可以避免死锁。

3.3. wait和notify通知机制

synchronized锁虽然可以解决多个线程竞争的情况，但无法解决多个线程合作协调的情况。

假设有一个队列类，期望的目标是

  • 线程1调用addTask不定期添加任务
  • 线程2调用getTask处理任务，如果没有任务，则会等待直到队里中有任务为止

这是一个典型的生产-消费者模式

class TaskQueue {
    Queue<String> queue = new LinkedList<>();

    public synchronized void addTask(String s) {
        this.queue.add(s);
    }

    public synchronized String getTask() throws InterruptedException {
        while (queue.isEmpty()) {
          // 等待获取任务
        }
        return queue.remove();
    }
}

由于getTask会获取this锁，因此当调用getTask时，其他线程是无法再调用addTask添加任务的，这就导致了一个死循环。

解决办法是调用锁对象上面的wait()方法，wait()是定义在Object对象上的native方法。

class TaskQueue {
        // ...
    public synchronized String getTask() throws InterruptedException {
        while (queue.isEmpty()) {
          this.wait()
        }
        return queue.remove();
    }
}

当调用wait之后，线程1就会释放this对象锁，这样其他线程就可以拿到this锁去执行addTask了。

当其他线程执行addTask之后，需要恢复线程1的getTask，这时候就需要调用this锁的notify了

class TaskQueue {
    Queue<String> queue = new LinkedList<>();

    public synchronized void addTask(String s) {
        this.queue.add(s);
        // this.notify();
        this.notifyAll();
    }
}

notify()只会唤醒某一个wiat()的线程，notifyAll()会唤起所有等待的线程，一般来说notifyAll更安全一些。

下面是完整的代码

import java.util.*;

public class Main {
    public static void main(String[] args) throws InterruptedException {
        var q = new TaskQueue();
          // 线程1：处理任务队列中的任务
        var handleThread = new Thread() {
            public void run() {
                while (true) {
                    try {
                        String s = q.getTask();
                        System.out.println("execute task: " + s);
                    } catch (InterruptedException e) {
                        return;
                    }
                }
            }
        };
        handleThread.start();
          // 线程2：添加任务
        var addThread = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                // 放入task:
                String s = "t-" + i;
                System.out.println("add task: " + s);
                q.addTask(s);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                }
            }
        });
        addThread.start();
        addThread.join();
        Thread.sleep(100);
          // 由于线程1是一个持久的循环任务，退出则需要直接中断
        handleThread.interrupt();
    }
}

class TaskQueue {
    Queue<String> queue = new LinkedList<>();

    public synchronized void addTask(String s) {
        this.queue.add(s);
        this.notifyAll();
    }

    public synchronized String getTask() throws InterruptedException {
          // 注意这里不能是 if
        while (queue.isEmpty()) {
            this.wait();
        }
        return queue.remove();
    }
}

3.4. 其他

此外还有信号量、管道等方法用于线程通信。这里暂不展开

4. 小结

本文整理了Java中多线程编程的基本开发，后续会进一步学习和更新。

由于对于Java语法和多线程编程都不是很熟悉，如果本文有错误，欢迎指正。

世人的悲欢并不相通，我只是觉得他们吵闹。

Copyright © Shymean 2016 - 2024蜀ICP备2023029820号