Java并发(二)-并发基础

Java

Java线程

现代操作系统调度的最小单位是线程，也叫轻量级进程(Light Weight Process)，在一个进程里可以创建多个线程，这些线程都拥有各自的计数器、堆栈和局部变量等属性，并且能够访问共享的内存变量。Java中线程由JVM管理。在HotSpot VM中，Java中的线程与操作系统中的线程一一对应。

线程的状态

Java线程在运行期间的生命周期主要有以下状态：

状态 说明
NEW 初始状态，线程被创建，但是还没有调用start()方法
RUNNABLE 运行状态，Java线程讲操作系统中的就绪和运行两种状态成为“运行中”
BLOCKED 阻塞状态，表示线程阻塞于锁
WAITING 等待状态，表示线程进入等待状态，进入该状态表示当前线程需要等待其他线程做出一些特定的动作(通知或中断)
TIME_WAITING 超时等待状态，该状态不同于WAITING状态于它可以在指定时间自行返回
TERMINATED 中止状态，表示当前线程已经执行完毕

Java中线程状态转换如下图(注：图来自《Java并发编程的艺术》)：

注：阻塞状态是线程阻塞在进入synchronized关键字修饰的方法或代码块是的状态，但是阻塞在concurrent包中的Lock接口的线程状态是等待状态，因为Lock包中的接口的实现依赖于LockSupport类中的相关方法。

Daemon线程

Daemon线程又叫后台线程，它主要被用作程序后台调度以及支持性工作，如：Java中的GC线程就是Daemon线程。在Java中可以调用setDaemon()方法把线程设置为当前线程的Daemon线程，如果前台线程执行完毕，后台线程会直接退出。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public class DaemonThread {

  public static void main(String[] args) throws InterruptedException {
    Thread thread = new Thread(() -> {
      try {
        TimeUnit.SECONDS.sleep(2);
      } catch (InterruptedException e) {
        e.printStackTrace();
      } finally {
        System.out.println("Daemon Thread Finally");
      }
    });
    thread.setDaemon(false);
    TimeUnit.SECONDS.sleep(1);
    System.out.println("Main Thread Finally");
  }
  
  Console Output:
  Main Thread Finally
}

线程中断

线程终止

在一些线程可能需要持久化的运行下去或者有条件的运行下去，我们可以使用thread.interupt()方法进行终止进行或者使用daemon线程让其随主线程同时终止，但这两种方式都不太美观，我们可以使用一个volatile boolean变量进行终止线程，主要使用到了volatile变量的内存可见性和单个操作的原子性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public class StopThread {

  public static void main(String[] args) throws InterruptedException {
    Task task = new Task();
    new Thread(task).start();
    TimeUnit.MILLISECONDS.sleep(500);
    task.stop();
    System.out.println("Main Stop");
  }

  static final class Task implements Runnable {

    private volatile boolean isStop;

    @Override
    public void run() {
      while (!isStop) {
        // 执行任务
        System.out.println("Task Running");
      }
      System.out.println("Task Done");
    }

    public void stop() {
      isStop = true;
    }
  }
}

我们可以看到这里通过isStop变量进行关闭线程执行，主要使用到了volatile变量的内存可见性。

线程间通信

Java中的线程通信主要以共享内存+信号量的方式进行隐式通信，Java也提供管道(Piple)进行线程间通信。Java中通过wait/notify机制实现共享内存+信号量的方式进行通信，Java中wait/notify主要有两种方式实现：synchronized+object、lock+condition，下面以生产者/消费者模型介绍以上两种实现：

synchronized+object

synchronized关键字可以修饰方法或者代码块，它能确保多个线程在同一时刻，只能有一个线程处于方法或者代码块中，它保证可线程对变量访问的内存可见性以及互斥性。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@Data
public class Factory {

  Product product;

  private final Producer producer = new Producer(this);

  private final Consumer consumer = new Consumer(this);

  public static void main(String[] args) throws InterruptedException {
    Factory factory = new Factory();
    Thread producerThread = new Thread(factory.getProducer());
    Thread consumerThread = new Thread(factory.getConsumer());
    producerThread.start();
    consumerThread.start();
    producerThread.join();
  }
  
  // 消费者
  static final class Consumer implements Runnable {

    private final Factory factory;

    Consumer(Factory factory) {
      this.factory = factory;
    }

    @Override
    public void run() {
      while (!Thread.interrupted()) {
        try {
          synchronized (factory.getConsumer()) {
            while (factory.product == null) {
              factory.getConsumer().wait();
            }
            TimeUnit.MILLISECONDS.sleep(500);
            System.out.println("Consumer consume: " + factory.product);
          }
          synchronized (factory.getProducer()) {
            factory.product = null;
            factory.getProducer().notifyAll();
          }
        } catch (InterruptedException e) {
          break;
        }
      }
    }
  }
  // 生产者
  static final class Producer implements Runnable {

    private final Factory factory;

    Producer(Factory factory) {
      this.factory = factory;
    }

    @Override
    public void run() {
      while (!Thread.interrupted()) {
        try {
          synchronized (factory.getConsumer()) {
            TimeUnit.MILLISECONDS.sleep(500);
            factory.product = new Product(JMockData.mock(Long.class));
            System.out.println("Producer product :" + factory.product);
            factory.getConsumer().notifyAll();
          }
          synchronized (factory.getProducer()) {
            while (factory.product != null) {
              factory.getProducer().wait();
            }
          }
        } catch (InterruptedException e) {
          break;
        }
      }
    }
  }
  // 产品
  @Data
  @AllArgsConstructor
  static final class Product {

    private Long id;

  }
}

上诉是等待/通知机制的实现，下图是等待/通知的状态转换示意图(注：图来自《Java并发编程的艺术》)

下面是等待/通知的经典写法：

等待方(消费者)

1
2
3
4
5
6
synchronized (对象) {
  while (条件) {
    对象.wait()；
  }
  对应的处理逻辑；
}

通知方(生产者)

1
2
3
4
synchronized (对象) {
  改变条件；
  对象.notifyAll()；
}

Lock+Condition

同样可以使用Lock+Condition来使用消息间通信。上诉使用synchronized+object使用的是Java中原生语法的锁对象，同时每个对象都维持一个同步队列，而使用Java util包中的lock对象可以比synchronized更多方式进行控制锁，同时需要condition维持同步队列。下面是使用lock+condition实现的上诉程序

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
public class LockFactory {

  final ReentrantLock lock = new ReentrantLock();

  final Condition produceCondition = lock.newCondition();

  final Condition consumeCondition = lock.newCondition();

  Product product;

  public static void main(String[] args) throws InterruptedException {
    LockFactory factory = new LockFactory();
    Thread produceThread = new Thread(new Producer(factory));
    Thread consumeThread = new Thread(new Consumer(factory));
    produceThread.start();
    consumeThread.start();
    consumeThread.join();
  }

  private static final class Producer implements Runnable {

    private final LockFactory factory;

    private Producer(LockFactory factory) {
      this.factory = factory;
    }

    @Override
    public void run() {
      while (!Thread.interrupted()) {
        factory.lock.lock();
        try {
          while (factory.product != null) {
            factory.produceCondition.await();
          }
          TimeUnit.MILLISECONDS.sleep(500);
          factory.product = new Product(JMockData.mock(long.class));
          System.out.println("Producer produce: " + factory.product);
          factory.consumeCondition.signal();
        } catch (InterruptedException e) {
          break;
        } finally {
          factory.lock.unlock();
        }
      }
    }
  }

  private static final class Consumer implements Runnable {

    private final LockFactory factory;

    private Consumer(LockFactory factory) {
      this.factory = factory;
    }

    @Override
    public void run() {
      while (!Thread.interrupted()) {
        factory.lock.lock();
        try {
          while (factory.product == null) {
            factory.consumeCondition.await();
          }
          TimeUnit.MILLISECONDS.sleep(500);
          System.out.println("Consumer consume: " + factory.product);
          factory.product = null;
          factory.produceCondition.signal();
        } catch (InterruptedException e) {
          break;
        } finally {
          factory.lock.unlock();
        }
      }
    }
  }

  @Data
  @AllArgsConstructor
  private static final class Product {

    private Long id;
  }
}

ThreadLocal

ThreadLocal从名字中可以看出指的是线程局部变量