线程间的通信

线程间的通信

volatile 和 synchronized 关键字

volatile关键字

volatile关键字修饰就是告知程序任何对变量的访问均需要从共享内存中获取,而对它的改变必须同步刷新到共享内存,保证所有线程对变量访问的可见性。

synchronized 关键字

synchronized关键字可以修饰方法或者以同步块的形式进行使用,确保多个线程在同一时刻,只能有一个线程处于方法或者同步块中,保证了线程对变量访问的可见性和排他性。

在下面的代码中,使用了同步块和同步方法,通过使用javap工具查看生成的class文件信息来分析synchronized关键字的实现细节,示例如下。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public class Synchronized {  
public static void main(String[] args) {

// 对Synchronized Class 对象加锁
synchronized (Synchronized.class) {

}
// 静态同步方法,对synchronized Clas加锁
m();
}

public static synchronized void m() {

}
}

在Synchronized.class同级目录执行javap -vSynchronized.class,部分相关输出如下所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public static void main(java.lang.String[]);  
// 方法修饰符,表示:public staticflags: ACC_PUBLIC, ACC_STATIC
Code:
stack=2, locals=1, args_size=1
0: ldc #1  // class com/murdock/books/multithread/book/Synchronized
2: dup
3: monitorenter  // monitorenter:监视器进入,获取锁
4: monitorexit   // monitorexit:监视器退出,释放锁
5: invokestatic  #16 // Method m:()V
8: return
public static synchronized void m();
// 方法修饰符,表示: public static synchronized
flags: ACC_PUBLIC, ACC_STATIC, ACC_SYNCHRONIZED
Code:
stack=0, locals=0, args_size=0
0: return

上面class信息中,对于同步块的实现使用了monitorenter和monitorexit指令,而同步方法则是依靠方法修饰符上的ACC_SYNCHRONIZED来完成的。无论采用哪种方式,其本质是对一个对象的监视器(monitor)进行获取,而这个获取过程是排他的,也就是同一时刻只能有一个线程获取到由synchronized所保护对象的监视器。
任意一个对象都拥有自己的监视器,当这个对象由同步块或者这个对象的同步方法调用时,执行方法的线程必须先获取到该对象的监视器才能进入同步块或者同步方法,而没有获取到监视器(执行该方法)的线程将会被阻塞在同步块和同步方法的入口处,进入BLOCKED状态。
下图描述了对象、对象的监视器、同步队列和执行线程之间的关系。
1

等待/通知机制

一个线程修改了一个对象的值,而另一个线程感知到了变化,然后进行相应的操作,整个过程开始于一个线程,而最终执行又是另一个线程。前者是生产者,后者就是消费者,这种模式隔离了“做什么”(what)和“怎么做”(How),在功能层面上实现了解耦,体系结构上具备了良好的伸缩性,但是在Java语言中如何实现类似的功能呢?
简单的办法是让消费者线程不断地循环检查变量是否符合预期,如下面代码所示,在while循环中设置不满足的条件,如果条件满足则退出while循环,从而完成消费者的工作。

1
2
3
4
while (value != desire) {  
Thread.sleep(1000);
}
doSomething();

上面这段伪代码在条件不满足时就睡眠一段时间,这样做的目的是防止过快的“无效”尝试,这种方式看似能够解实现所需的功能,但是却存在如下问题。

  1. 难以确保及时性。在睡眠时,基本不消耗处理器资源,但是如果睡得过久,就不能及时发现条件已经变化,也就是及时性难以保证。
  2. 难以降低开销。如果降低睡眠的时间,比如休眠1毫秒,这样消费者能更加迅速地发现条件变化,但是却可能消耗更多的处理器资源,造成了无端的浪费。

以上两个问题,看似矛盾难以调和,但是Java通过内置的等待/通知机制能够很好地解决这个矛盾并实现所需的功能。

方法名 描述
notify() 通知一个在对象上等待的线程,使其从wait()方法返回,而返回的前提是该线程获取到了对象的锁
notifyAll() 通知所有等待在该对象上的线程
wait() 调用该方法的线程进入WAITING状态,只有等待另外线程的通知或被中断才返回,需要注意,电泳wait()方法后,会释放对象的锁
wait(long) 超时等待一段时间,这里的参数时间是毫秒,也就是等待长待n毫秒,如果没有通知就超时返回
wait(long, int) 对一超时时间更细粒度的控制,可以达到纳秒

等待/通知机制,是指一个线程A调用了对象O的wait()方法进入等待状态,而另一个线程B调用了对象O的notify()或者notifyAll()方法,线程A收到通知后从对象O的wait()方法返回,进而执行后续操作。上述两个线程通过对象O来完成交互,而对象上的wait()和notify/notifyAll()的关系就如同开关信号一样,用来完成等待方和通知方之间的交互工作。
在代码清单4-11所示的例子中,创建了两个线程——WaitThread和NotifyThread,前者检查flag值是否为false,如果符合要求,进行后续操作,否则在lock上等待,后者在睡眠了一段时间后对lock进行通知,示例如下所示。

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
public class WaitNotify {  
static boolean flag = true;
static Object lock = new Object();

public static void main(String[] args) throws Exception {
Thread waitThread = new Thread(new Wait(), "WaitThread");
waitThread.start();
TimeUnit.SECONDS.sleep(1);
Thread notifyThread = new Thread(new Notify(), "NotifyThread");
notifyThread.start();
}

static class Wait implements Runnable {
public void run() {
// 加锁,拥有lock的Monitor
synchronized (lock) {
// 当条件不满足时,继续wait,同时释放了lock的锁
while (flag) {
try {
System.out.println(Thread.currentThread()
+ " flag is true. wait@ "
+ new SimpleDateFormat("HH:mm:ss")
.format(new Date()));
lock.wait();
} catch (InterruptedException e) {
}
}
// 条件满足时,完成工作
System.out.println(Thread.currentThread()
+ " flag is false. running@ "
+ new SimpleDateFormat("HH:mm:ss").format(new Date()));
}
}
}

static class Notify implements Runnable {
public void run() {
// 加锁,拥有lock的Monitor
synchronized (lock) {
// 获取lock的锁,然后进行通知,通知时不会释放lock的锁,
// 直到当前线程释放了lock后,WaitThread才能从wait方法中返回
System.out.println(Thread.currentThread()
+ " hold lock. notify @ "
+ new SimpleDateFormat("HH:mm:ss").format(new Date()));
lock.notifyAll();
flag = false;
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 再次加锁
synchronized (lock) {
System.out.println(Thread.currentThread()
+ " hold lock again. sleep@ "
+ new SimpleDateFormat("HH:mm:ss").format(new Date()));
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}

输出如下(输出内容可能不同,主要区别在时间上)

1
2
3
4
Thread[WaitThread,5,main] flag is true. wait@ 16:15:44  
Thread[NotifyThread,5,main] hold lock. notify @ 16:15:45
Thread[NotifyThread,5,main] hold lock again. sleep@ 16:15:50
Thread[WaitThread,5,main] flag is false. running@ 16:15:55

上述例子主要说明了调用wait()、notify()以及notifyAll()时需要注意的细节,如下。

  1. 使用wait()、notify()和notifyAll()时需要先对调用对象加锁。
  2. 调用wait()方法后,线程状态由RUNNING变为WAITING,并将当前线程放置到对象的等待队列。
  3. notify()或notifyAll()方法调用后,等待线程依旧不会从wait()返回,需要调用notify()或notifAll()的线程释放锁之后,等待线程才有机会从wait()返回。
  4. notify()方法将等待队列中的一个等待线程从等待队列中移到同步队列中,而notifyAll()方法则是将等待队列中所有的线程全部移到同步队列,被移动的线程状态由WAITING变为BLOCKED。
  5. 从wait()方法返回的前提是获得了调用对象的锁。

从上述细节中可以看到,等待/通知机制依托于同步机制,其目的就是确保等待线程从wait()方法返回时能够感知到通知线程对变量做出的修改。下图描述了上述示例的过程。
2

在图中,WaitThread首先获取了对象的锁,然后调用对象的wait()方法,从而放弃了锁并进入了对象的等待队列WaitQueue中,进入等待状态。由于WaitThread释放了对象的锁,NotifyThread随后获取了对象的锁,并调用对象的notify()方法,将WaitThread从WaitQueue移到SynchronizedQueue中,此时WaitThread的状态变为阻塞状态。NotifyThread释放了锁之后,WaitThread再次获取到锁并从wait()方法返回继续执行。

等待/通知机制的经典范式—消费者/生产者模式

等待方(消费者)遵循原则:

  1. 获取对象的锁
  2. 如果条件不满足,那么调用对象的wait()方法,被通知后仍要检查条件
  3. 条件满足则执行对应的逻辑
1
2
3
4
5
6
7
//伪代码实现
synchronized(对象) {
while(条件不满足) {
对象.wait();
}
对应的处理逻辑
}

通知方(生产者)遵循原则:

  1. 获得对象的锁
  2. 改变条件
  3. 通知所有等待在对象上的线程
1
2
3
4
5
//伪代码实现
synchronized(对象) {
改变条件
对象.notifyAll();
}

管道输入/输出流

管道输入/输出流和普通的文件输入/输出流或者网络输入/输出流不同之处在于,它主要用于线程之间的数据传输,而传输媒介为内存。
管道输入/输出流主要包括4中具体实现:

  • PipedOutputStream/PipedInputStream:面向字节
  • PipedReader/PipedWriter:面向字符

对于Piped类型的流,必须先要进行绑定,也就是调用connect()方法,如果没有将输入/输出流绑定起来,对于该流的访问将会抛出异常。

Thread.join()的使用