Paincker

Java线程相关基础知识梳理

发表于 分类于 阅读次数： Valine： 本文字数： 8.9k 阅读时长 ≈ 8 分钟

Thread

状态

  • New：创建但是未启动。

    A thread that has not yet started is in this state.

  • Runnable：可运行。包含Ready和Running两种子状态，在线程调度时来回切换。

    A thread executing in the Java virtual machine is in this state.

  • Blocked：请求并等待锁。

    A thread that is blocked waiting for a monitor lock is in this state.

  • Waiting：等待其他线程的Notification。

    A thread that is waiting indefinitely for another thread to perform a particular action is in this state.

  • Timed Waiting：等待其他线程的Notification并设置了超时。

    A thread that is waiting for another thread to perform an action for up to a specified waiting time is in this state.

  • Terminated：执行结束。

    A thread that has exited is in this state.

状态转移图

图片来源：https://www.geeksforgeeks.org/lifecycle-and-states-of-a-thread-in-java/

图片来源：https://blog.csdn.net/m0_37779570/article/details/84938476

stop和interrupt

Thread有stop和interrupt方法，都是用于终止线程（转移到Terminated状态）。

  • stop：强制终止这个线程。线程会停止执行，加的锁也都会被释放，容易导致问题，因此已被标记为Deprecated。

  • interrupt：设置线程的中断标志为true，具体是否会中断要看线程自身的处理。另外，当线程处于阻塞状态时，调用interrupt会抛出InterruptedException异常并清除中断标志，将线程唤醒。

1
2
3
4
5
6
7
8
9
10
11
12
public void run() {
    try {
        // 判断是否被中断，以及还有没有任务，决定要不要继续运行
        while (!Thread.currentThread().isInterrupted() && moreWorkToDo()) {
            doMoreWork();
        }
    } catch (InterruptedException e) {
        // 线程在wait或sleep期间被中断了
    } finally {
        // 线程结束前做一些清理工作
    }
}

线程被允许调用自身的stop和interrupt方法。而在一个线程中调用其他Thread对象的stop和interrupt方法时，SecurityManager会检查权限，有可能会抛出SecurityException异常。

参考：https://www.cnblogs.com/onlywujun/p/3565082.html

UncaughtExceptionHandler

未捕获异常的Handler

  • Thread.setDefaultUncaughtExceptionHandler(handler)：给所有线程设置
  • someThread.setUncaughtExceptionHandler(handler)：给特定线程设置

线程相关API简介

Runnable与Callable

  • Runnable: 只有一个 void run(); 方法，没有返回值
  • Callable：只有一个 V call() throws Exception; 方法，有泛型返回值，可能会抛异常
1
2
3
public interface Runnable {
    void run();
}
1
2
3
public interface Callable<V> {
    V call() throws Exception;
}

Future

代表一个异步计算的结果

  • get：等待并获取计算结果，有无限等待和设置超时两个版本。
  • cancel：取消任务
1
2
3
4
5
6
7
8
public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

Executor

执行Task

1
2
3
public interface Executor {
    void execute(Runnable command);
}

ExecutorService

继承自Executor，执行Task，并可以提供更复杂的控制，包括取消任务、终止服务 (shutdown) 等。

  • shutdown：禁止提交新任务，允许已提交任务继续执行。
  • awaitTermination：等待已提交任务执行结束（可设置超时），阻塞方法。
  • shutdownNow：尝试终止已提交任务。
  • submit：提交Runnable或Callable，返回Future。
  • invokeAll：提交多个任务，返回Future的List。
  • invokeAny：提交多个任务，返回第一个执行成功的任务的结果，可以设置超时。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public interface ExecutorService extends Executor {
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;

    <T> Future<T> submit(Callable<T> task);
    <T> Future<T> submit(Runnable task, T result);
    Future<?> submit(Runnable task);

    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException;
    <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;
    <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

ThreadPoolExecutor

ExecutorService接口最常见的一个实现。

使用线程池的中某个线程执行提交的任务。解决两个问题：大量异步任务的高效率执行；管理线程资源。

  • corePoolSize：核心线程数，存活的最小线程数量
  • maximumPoolSize：最大线程数
  • workQueue：BlockingQueue<Runnable>，工作队列
  • keepAliveTime：超过最小数量的线程空闲等待时间，超时后线程退出
  • threadFactory：创建线程的工厂
  • handler：RejectedExecutionHandler，任务提交失败时调用（线程池饱和或者被Shutdown）

ForkJoinPool

ExecutorService接口的一个实现。

Executors.newWorkStealingPool()返回一个ForkJoinPool。工作窃取算法允许已经耗尽输入队列中的工作项的线程从其他队列“窃取”工作项。目标是在处理器之间分配工作项，从而最大限度地利用所有可用的处理器来完成计算密集型任务。

Executors

创建线程池的工厂。

ThreadPoolExecutor队列设计

队列为BlockingQueue类型，用于保存提交的任务。

添加新任务时，按照CoreThread -> BlockingQueue -> NormalThread的优先级处理：

  • 如果运行的线程数少于corePoolSize，添加新任务会启动一个新线程。
  • 如果运行的线程数大于等于corePoolSize，添加新任务会先尝试添加到工作队列中。
  • 如果工作队列添加失败，则尝试创建新线程，如果线程超过maximumPoolSize，任务会被拒绝。

JavaDoc原文：

Any BlockingQueue may be used to transfer and hold submitted tasks. The use of this queue interacts with pool sizing:

  • If fewer than corePoolSize threads are running, the Executor always prefers adding a new thread rather than queuing.
  • If corePoolSize or more threads are running, the Executor always prefers queuing a request rather than adding a new thread.
  • If a request cannot be queued, a new thread is created unless this would exceed maximumPoolSize, in which case, the task will be rejected.

有三种队列策略：

  • SynchronousQueue，队列本身不保存数据，添加后必须有空闲线程立即执行。
  • LinkedBlockingQueue，队列保存所有提交的任务，因此线程池只有核心线程。
  • ArrayBlockingQueue，队列有固定容量，可以保存一定数量的任务。

JavaDoc原文：

There are three general strategies for queuing:

  • Direct handoffs. A good default choice for a work queue is a SynchronousQueue that hands off tasks to threads without otherwise holding them. Here, an attempt to queue a task will fail if no threads are immediately available to run it, so a new thread will be constructed. This policy avoids lockups when handling sets of requests that might have internal dependencies. Direct handoffs generally require unbounded maximumPoolSizes to avoid rejection of new submitted tasks. This in turn admits the possibility of unbounded thread growth when commands continue to arrive on average faster than they can be processed.
  • Unbounded queues. Using an unbounded queue (for example a LinkedBlockingQueue without a predefined capacity) will cause new tasks to wait in the queue when all corePoolSize threads are busy. Thus, no more than corePoolSize threads will ever be created. (And the value of the maximumPoolSize therefore doesn’t have any effect.) This may be appropriate when each task is completely independent of others, so tasks cannot affect each others execution; for example, in a web page server. While this style of queuing can be useful in smoothing out transient bursts of requests, it admits the possibility of unbounded work queue growth when commands continue to arrive on average faster than they can be processed.
  • Bounded queues. A bounded queue (for example, an ArrayBlockingQueue) helps prevent resource exhaustion when used with finite maximumPoolSizes, but can be more difficult to tune and control. Queue sizes and maximum pool sizes may be traded off for each other: Using large queues and small pools minimizes CPU usage, OS resources, and context-switching overhead, but can lead to artificially low throughput. If tasks frequently block (for example if they are I/O bound), a system may be able to schedule time for more threads than you otherwise allow. Use of small queues generally requires larger pool sizes, which keeps CPUs busier but may encounter unacceptable scheduling overhead, which also decreases throughput.

synchronized关键字与ReentrantLock

用法

1
2
3
4
5
public synchronized void test() {
}

synchronized（someObject） {
}
1
2
3
4
5
6
7
8
9
10
11
ReentrantLock lock = new ReentrantLock();
public void run() {
    // lock要写在try外面。如果lock失败，不会执行finally中的unlock
    lock.lock();
    try {
        // ...
    } finally {
        // unlock要写在finally中。无论执行是否异常，都要确保unlock执行
        lock.unlock();
    }
}

相同点

  • 都是通过加锁阻塞的方式实现线程同步。
  • 都是可重入锁。

区别

ReentrantLock功能更多。

  • synchronized是关键字，使用更简单，ReentrantLock需要按照固定格式使用。
  • synchronized可以修饰方法、代码块，ReentrantLock只能作用于代码块。
  • synchronized等待不可中断，ReentrantLock可以。
  • synchronized是非公平锁，ReentrantLock默认非公平，但可以设置成公平锁。
  • 一个synchronized只能实现一个条件，ReentrantLock可以绑定多个Condition。

重入锁和不可重入锁

重入锁可以递归调用（或者是两个代码块用了相同的锁，可以嵌套调用，如下代码示意），而非重入锁不能递归调用。

1
2
3
4
5
6
7
8
9
10
11
public void method1() {
    synchronized (SynchronizedTest.class) {
        System.out.println("1");
        method2();
    }
}
public void method2() {
    synchronized (SynchronizedTest.class) {
        System.out.println("2");
    }
}

公平锁与非公平锁

  • 公平锁：锁释放时，按照申请时间顺序依次获得锁。
  • 非公平锁：锁释放时，任意等待线程随机获得锁。

参考：

https://www.jianshu.com/p/96c89e6e7e90

https://juejin.im/post/5bc87409f265da0ad701da35

https://blog.csdn.net/fuyuwei2015/article/details/83387536

volatile关键字

使用volatile的三个理由

  • 字分裂。64位的long 和 double 类型，可能会在两个单独的32位操作中执行。
  • 保证内存可见性。防止编译器优化，读写直接在内存进行，而不会复制变量到CPU寄存器。一旦该字段发生写操作，所有任务的读操作都将看到更改。
  • 防止指令重排。

单例模式中的 volatile

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public class Singleton {
    private static volatile Singleton instance = null; // 需要volatile修饰
    public static Singleton getInstance() {
        if (instance == null) {
            synchronzied (Singleton.class) {
                if (instance == null) {
                    instance = new Singleton();  // 非原子操作
                }
            }
        }
        return instance;
    }
    private Singleton() { }
}

语句instance = new Singleton() 由三步组成：

1
2
3
memory = allocate();   // 1：分配对象的内存空间
ctorInstance(memory);  // 2：初始化对象
instance = memory;     // 3：设置instance指向刚分配的内存地址

如果没有volatile修饰，可能发生重排，变成

1
2
3
memory = allocate();   // 1：分配对象的内存空间
instance = memory;     // 3：instance指向刚分配的内存地址，此时对象还未初始化
ctorInstance(memory);  // 2：初始化对象

执行3赋值后，但是还没执行2初始化，此时如果切换到了另一个线程，发现instance已经赋值，就会直接使用未初始化的对象。

ThreadLocal

每个线程有一个对象的副本，线程之间访问的变量是隔离的。

实现

有两种实现思路：

  • ThreadLocal内包含一个Map，用线程作为Key，Value为值。这样写Map需要保证线程安全，要用锁，性能较差。
  • 【实际使用的实现】每个Thread对象里本来就有一个Map，ThreadLocal本身不存数据，只是作为Key使用，Value为值。
1
2
3
4
5
6
7
8
9
ThreadLocal<String> t = new ThreadLocal<>();

// 读写当前线程的副本
t.get();
t.set("xxx");

// 实际上等效于下面的代码（但是因为threadLocals变量为包权限，外部没法直接访问）
Thread.currentThread().threadLocals.get(t);
Thread.currentThread().threadLocals.set(t, "xxx");

实际案例

  • 数据库的Connection，每个线程只操作自己线程独立的Connection
  • Java Web后台应用，每个线程有自己单独的 Session 实例
  • Android中每个线程有自己的Looper，不会重复创建
相关文章