Java并发

一、使用线程

有三种使用线程的方法:

  • 实现 Runnable 接口;
  • 实现 Callable 接口;
  • 继承 Thread 类。

实现 Runnable 和 Callable 接口的类只能当做一个可以在线程中运行的任务,不是真正意义上的线程,因此最后还需要通过 Thread 来调用。可以说任务是通过线程驱动从而执行的。

实现 Runnable 接口

需要实现 run() 方法。

通过 Thread 调用 start() 方法来启动线程。

  1. public class MyRunnable implements Runnable {
  2. public void run() {
  3. // ...
  4. }
  5. }
  1. public static void main(String[] args) {
  2. MyRunnable instance = new MyRunnable();
  3. Thread thread = new Thread(instance);
  4. thread.start();
  5. }

实现 Callable 接口

与 Runnable 相比,Callable 可以有返回值,返回值通过 FutureTask 进行封装。

  1. public class MyCallable implements Callable<Integer> {
  2. public Integer call() {
  3. return 123;
  4. }
  5. }
  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2. MyCallable mc = new MyCallable();
  3. FutureTask<Integer> ft = new FutureTask<>(mc);
  4. Thread thread = new Thread(ft);
  5. thread.start();
  6. System.out.println(ft.get());
  7. }

继承 Thread 类

同样也是需要实现 run() 方法,因为 Thread 类也实现了 Runable 接口。

当调用 start() 方法启动一个线程时,虚拟机会将该线程放入就绪队列中等待被调度,当一个线程被调度时会执行该线程的 run() 方法。

  1. public class MyThread extends Thread {
  2. public void run() {
  3. // ...
  4. }
  5. }
  1. public static void main(String[] args) {
  2. MyThread mt = new MyThread();
  3. mt.start();
  4. }

实现接口 VS 继承 Thread

实现接口会更好一些,因为:

  • Java 不支持多重继承,因此继承了 Thread 类就无法继承其它类,但是可以实现多个接口;
  • 类可能只要求可执行就行,继承整个 Thread 类开销过大。

二、基础线程机制

Executor

Executor 管理多个异步任务的执行,而无需程序员显式地管理线程的生命周期。这里的异步是指多个任务的执行互不干扰,不需要进行同步操作。

主要有三种 Executor:

  • CachedThreadPool:一个任务创建一个线程;
  • FixedThreadPool:所有任务只能使用固定大小的线程;
  • SingleThreadExecutor:相当于大小为 1 的 FixedThreadPool。
  1. public static void main(String[] args) {
  2. ExecutorService executorService = Executors.newCachedThreadPool();
  3. for (int i = 0; i < 5; i++) {
  4. executorService.execute(new MyRunnable());
  5. }
  6. executorService.shutdown();
  7. }

Daemon

守护线程是程序运行时在后台提供服务的线程,不属于程序中不可或缺的部分。

当所有非守护线程结束时,程序也就终止,同时会杀死所有守护线程。

main() 属于非守护线程。

在线程启动之前使用 setDaemon() 方法可以将一个线程设置为守护线程。

  1. public static void main(String[] args) {
  2. Thread thread = new Thread(new MyRunnable());
  3. thread.setDaemon(true);
  4. }

sleep()

Thread.sleep(millisec) 方法会休眠当前正在执行的线程,millisec 单位为毫秒。

sleep() 可能会抛出 InterruptedException,因为异常不能跨线程传播回 main() 中,因此必须在本地进行处理。线程中抛出的其它异常也同样需要在本地进行处理。

  1. public void run() {
  2. try {
  3. Thread.sleep(3000);
  4. } catch (InterruptedException e) {
  5. e.printStackTrace();
  6. }
  7. }

yield()

对静态方法 Thread.yield() 的调用声明了当前线程已经完成了生命周期中最重要的部分,可以切换给其它线程来执行。该方法只是对线程调度器的一个建议,而且也只是建议具有相同优先级的其它线程可以运行。

  1. public void run() {
  2. Thread.yield();
  3. }

三、中断

一个线程执行完毕之后会自动结束,如果在运行过程中发生异常也会提前结束。

InterruptedException

通过调用一个线程的 interrupt() 来中断该线程,如果该线程处于阻塞、限期等待或者无限期等待状态,那么就会抛出 InterruptedException,从而提前结束该线程。但是不能中断 I/O 阻塞和 synchronized 锁阻塞。

对于以下代码,在 main() 中启动一个线程之后再中断它,由于线程中调用了 Thread.sleep() 方法,因此会抛出一个 InterruptedException,从而提前结束线程,不执行之后的语句。

  1. public class InterruptExample {
  2. private static class MyThread1 extends Thread {
  3. @Override
  4. public void run() {
  5. try {
  6. Thread.sleep(2000);
  7. System.out.println("Thread run");
  8. } catch (InterruptedException e) {
  9. e.printStackTrace();
  10. }
  11. }
  12. }
  13. }
  1. public static void main(String[] args) throws InterruptedException {
  2. Thread thread1 = new MyThread1();
  3. thread1.start();
  4. thread1.interrupt();
  5. System.out.println("Main run");
  6. }
  1. Main run
  2. java.lang.InterruptedException: sleep interrupted
  3. at java.lang.Thread.sleep(Native Method)
  4. at InterruptExample.lambda$main$0(InterruptExample.java:5)
  5. at InterruptExample$$Lambda$1/713338599.run(Unknown Source)
  6. at java.lang.Thread.run(Thread.java:745)

interrupted()

如果一个线程的 run() 方法执行一个无限循环,并且没有执行 sleep() 等会抛出 InterruptedException 的操作,那么调用线程的 interrupt() 方法就无法使线程提前结束。

但是调用 interrupt() 方法会设置线程的中断标记,此时调用 interrupted() 方法会返回 true。因此可以在循环体中使用 interrupted() 方法来判断线程是否处于中断状态,从而提前结束线程。

  1. public class InterruptExample {
  2. private static class MyThread2 extends Thread {
  3. @Override
  4. public void run() {
  5. while (!interrupted()) {
  6. // ..
  7. }
  8. System.out.println("Thread end");
  9. }
  10. }
  11. }
  1. public static void main(String[] args) throws InterruptedException {
  2. Thread thread2 = new MyThread2();
  3. thread2.start();
  4. thread2.interrupt();
  5. }
  1. Thread end

Executor 的中断操作

调用 Executor 的 shutdown() 方法会等待线程都执行完毕之后再关闭,但是如果调用的是 shutdownNow() 方法,则相当于调用每个线程的 interrupt() 方法。

以下使用 Lambda 创建线程,相当于创建了一个匿名内部线程。

  1. public static void main(String[] args) {
  2. ExecutorService executorService = Executors.newCachedThreadPool();
  3. executorService.execute(() -> {
  4. try {
  5. Thread.sleep(2000);
  6. System.out.println("Thread run");
  7. } catch (InterruptedException e) {
  8. e.printStackTrace();
  9. }
  10. });
  11. executorService.shutdownNow();
  12. System.out.println("Main run");
  13. }
  1. Main run
  2. java.lang.InterruptedException: sleep interrupted
  3. at java.lang.Thread.sleep(Native Method)
  4. at ExecutorInterruptExample.lambda$main$0(ExecutorInterruptExample.java:9)
  5. at ExecutorInterruptExample$$Lambda$1/1160460865.run(Unknown Source)
  6. at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  7. at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  8. at java.lang.Thread.run(Thread.java:745)

如果只想中断 Executor 中的一个线程,可以通过使用 submit() 方法来提交一个线程,它会返回一个 Future<?> 对象,通过调用该对象的 cancel(true) 方法就可以中断线程。

  1. Future<?> future = executorService.submit(() -> {
  2. // ..
  3. });
  4. future.cancel(true);

四、互斥同步

Java 提供了两种锁机制来控制多个线程对共享资源的互斥访问,第一个是 JVM 实现的 synchronized,而另一个是 JDK 实现的 ReentrantLock。

synchronized

1. 同步一个代码块

  1. public void func() {
  2. synchronized (this) {
  3. // ...
  4. }
  5. }

它只作用于同一个对象,如果调用两个对象上的同步代码块,就不会进行同步。

对于以下代码,使用 ExecutorService 执行了两个线程,由于调用的是同一个对象的同步代码块,因此这两个线程会进行同步,当一个线程进入同步语句块时,另一个线程就必须等待。

  1. public class SynchronizedExample {
  2. public void func1() {
  3. synchronized (this) {
  4. for (int i = 0; i < 10; i++) {
  5. System.out.print(i + " ");
  6. }
  7. }
  8. }
  9. }
  1. public static void main(String[] args) {
  2. SynchronizedExample e1 = new SynchronizedExample();
  3. ExecutorService executorService = Executors.newCachedThreadPool();
  4. executorService.execute(() -> e1.func1());
  5. executorService.execute(() -> e1.func1());
  6. }
  1. 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9

对于以下代码,两个线程调用了不同对象的同步代码块,因此这两个线程就不需要同步。从输出结果可以看出,两个线程交叉执行。

  1. public static void main(String[] args) {
  2. SynchronizedExample e1 = new SynchronizedExample();
  3. SynchronizedExample e2 = new SynchronizedExample();
  4. ExecutorService executorService = Executors.newCachedThreadPool();
  5. executorService.execute(() -> e1.func1());
  6. executorService.execute(() -> e2.func1());
  7. }
  1. 0 0 1 1 2 2 3 3 4 4 5 5 6 6 7 7 8 8 9 9

2. 同步一个方法

  1. public synchronized void func () {
  2. // ...
  3. }

它和同步代码块一样,作用于同一个对象。

3. 同步一个类

  1. public void func() {
  2. synchronized (SynchronizedExample.class) {
  3. // ...
  4. }
  5. }

作用于整个类,也就是说两个线程调用同一个类的不同对象上的这种同步语句,也会进行同步。

  1. public class SynchronizedExample {
  2. public void func2() {
  3. synchronized (SynchronizedExample.class) {
  4. for (int i = 0; i < 10; i++) {
  5. System.out.print(i + " ");
  6. }
  7. }
  8. }
  9. }
  1. public static void main(String[] args) {
  2. SynchronizedExample e1 = new SynchronizedExample();
  3. SynchronizedExample e2 = new SynchronizedExample();
  4. ExecutorService executorService = Executors.newCachedThreadPool();
  5. executorService.execute(() -> e1.func2());
  6. executorService.execute(() -> e2.func2());
  7. }
  1. 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9

4. 同步一个静态方法

  1. public synchronized static void fun() {
  2. // ...
  3. }

作用于整个类。

ReentrantLock

ReentrantLock 是 java.util.concurrent(J.U.C)包中的锁。

  1. public class LockExample {
  2. private Lock lock = new ReentrantLock();
  3. public void func() {
  4. lock.lock();
  5. try {
  6. for (int i = 0; i < 10; i++) {
  7. System.out.print(i + " ");
  8. }
  9. } finally {
  10. lock.unlock(); // 确保释放锁,从而避免发生死锁。
  11. }
  12. }
  13. }
  1. public static void main(String[] args) {
  2. LockExample lockExample = new LockExample();
  3. ExecutorService executorService = Executors.newCachedThreadPool();
  4. executorService.execute(() -> lockExample.func());
  5. executorService.execute(() -> lockExample.func());
  6. }
  1. 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9

一个线程在持有一个锁的时候,它内部能否再次(多次)申请该锁。如果一个线程已经获得了锁,其内部还可以多次申请该锁成功。那么我们就称该锁为可重入锁。

典型使用场景:

场景1:有状态执行(如果发现该操作已经在执行中则不再执行)

  1. private ReentrantLock lock = new ReentrantLock();
  2. ...
  3. //如果已经被lock,则立即返回false不会等待,达到忽略操作的效果
  4. if (lock.tryLock()) {
  5. try {
  6. //操作
  7. } finally {
  8. lock.unlock();
  9. }
  10. }

场景2:同步执行(如果发现该操作已经在执行,等待一个一个执行)
类似synchronized。但是,ReentrantLock有更灵活的锁定方式,支持公平锁(默认)与不公平锁,而synchronized永远是公平的。

  • 公平锁:操作会排一个队按顺序执行,来保证执行顺序。(会消耗更多的时间来排队)
  • 不公平锁:是无序状态允许插队,jvm会自动计算如何处理更快速来调度插队。(如果不关心顺序,这个速度会更快)
  1. //参数默认false,不公平锁
  2. private ReentrantLock lock = new ReentrantLock();
  3. //公平锁
  4. private ReentrantLock lock = new ReentrantLock(true);
  5. try {
  6. // 如果被其它资源锁定,会在此等待锁释放,达到暂停的效果
  7. lock.lock();
  8. //操作
  9. } finally {
  10. lock.unlock();
  11. }

场景3:尝试等待执行(如果发现该操作已经在执行,则尝试等待一段时间,等待超时则不执行)
这种其实属于场景2的改进,等待获得锁的操作有一个时间的限制,如果超时则放弃执行。
用来防止由于资源处理不当长时间占用导致死锁情况(大家都在等待资源,导致线程队列溢出)。

  1. try {
  2. // 如果已经被lock,尝试等待5s,看是否可以获得锁,如果5s后仍然无法获得锁则返回false继续执行
  3. if (lock.tryLock(5, TimeUnit.SECONDS)) {
  4. try {
  5. //操作
  6. } finally {
  7. lock.unlock();
  8. }
  9. }
  10. } catch (InterruptedException e) {
  11. // 当前线程被中断时(interrupt),会抛InterruptedException
  12. e.printStackTrace();
  13. }

场景4:可中断执行(如果发现该操作已经在执行,等待执行。这时可中断正在进行的操作立刻释放锁继续下一操作)
synchronized与Lock在默认情况下是不会响应中断(interrupt)操作,会继续执行完。
lockInterruptibly()提供了可中断锁来解决此问题。(场景2的另一种改进,没有超时,只能等待中断或执行完毕)
这种情况主要用于取消某些操作对资源的占用。如:(取消正在同步运行的操作,来防止不正常操作长时间占用造成的阻塞)

  1. try {
  2. lock.lockInterruptibly();
  3. //操作
  4. } catch (InterruptedException e) {
  5. e.printStackTrace();
  6. } finally {
  7. lock.unlock();
  8. }

比较

1. 锁的实现

synchronized 是 JVM 实现的,而 ReentrantLock 是 JDK 实现的。

2. 性能

新版本 Java 对 synchronized 进行了很多优化,例如自旋锁等,synchronized 与 ReentrantLock 大致相同。

3. 等待可中断

当持有锁的线程长期不释放锁的时候,正在等待的线程可以选择放弃等待,改为处理其他事情。

ReentrantLock 可中断,而 synchronized 不行。

4. 公平锁

公平锁是指多个线程在等待同一个锁时,必须按照申请锁的时间顺序来依次获得锁。

synchronized 中的锁是非公平的,ReentrantLock 默认情况下也是非公平的,但是也可以是公平的。

5. 锁绑定多个条件

一个 ReentrantLock 可以同时绑定多个 Condition 对象。

使用选择

除非需要使用 ReentrantLock 的高级功能,否则优先使用 synchronized。这是因为 synchronized 是 JVM 实现的一种锁机制,JVM 原生地支持它,而 ReentrantLock 不是所有的 JDK 版本都支持。