线程之间的通信又称为线程同步,是指当某个线程修改了一个对象的值时,另外一个线程能感知到该值的变化并进行相应的操作。
实现线程之间通信的方法有:
1.基于volatile修饰的共享变量
2.通过wait和notify机制
3.Thread.Join方法
4.使用synchronized同步关键字
5.Condition.await/signal方法
Java提供了(wait/notify)等待/通知机制来实现多个线程之间的协同处理,也就是控制线程之间的等待和唤醒。
wait()方法,使当前线程进入阻塞状态,并且释放持有的锁。
notify()方法,唤醒处于阻塞状态的下一个线程
notifyAll()方法,唤醒所有处于阻塞状态下的线程。
线程的通信(wait/notify)等待/通知机制,是指一个线程A调用了对象O的wait()方法进入等待状态,而另一个线程B调用了对象O的notify()或者notifyAll()方法,线程A收到通知后从对象O的wait()方法返回,进而执行后续操作。上述两个线程通过对象O来完成交互,而对象上的wait()和notify/notifyAll()的关系就如同开关信号一样,用来完成等待方和通知方之间的交互工作。
生产者/消费者是经典的线程通信与协同机制,接下来,我们设计一个简单的模型。
生产者:
- public class Product implements Runnable {
- private Queue
msg; - private int maxSize;
-
- public Product(Queue
msg, int maxSize ) { - this.msg = msg;
- this.maxSize = maxSize;
- }
- @Override
- public void run() {
- int i=0;
- while (true) {
- synchronized (msg) {
- while (msg.size() == maxSize) {
- //生产者满了
- try {
- msg.wait();
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
-
- System.out.println("生产...");
- msg.add("生产消息"+i);
- msg.notify();
- msg.notifyAll();
- }
- }
- }
- }
Producer表示一个生产者线程,该线程定义了一个共享对象msg,然后在run()方法中实现如下逻辑:
使用synchronized(msg)对共享对象加锁。
当msg.size==maxSize时,表示队列满了,让当前线程等待。
否则往bags中添加数据,并且使用msg.notify()方法唤醒阻塞的消费者线程。
从代码中可以发现,wait()和notify()方法必须要写在synchronized代码块中,至于原因 ,后面再分析。
消费者:
- public class Consumer implements Runnable {
- private Queue
msg; - private int maxSize;
-
- public Consumer(Queue
msg, int maxSize ) { - this.msg = msg;
- this.maxSize = maxSize;
- }
-
- @Override
- public void run() {
- while (true){
- synchronized (msg){
- //消费者空了
- while (msg.isEmpty()){
- try {
- msg.wait();//阻塞当前队列
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
-
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- System.out.println("消费者消费消息"+msg.remove());
- msg.notify();//唤醒处于等待状态的生产者
- }
- }
- }
- }
Consumer表示一个消费者线程,同样会用到共享对象msg,接着在run()方法中实现如下逻辑:
先使用synchronized(msg)方法加锁,注意生产者/消费者锁定的对象实例必须是同一个。
如果msg.isEmpty()表示队列空了,则需要等待生产者生产才能继续消费。
否则,使用msg.remove()方法从队列中消费数据,当数据消费了之后,可以使用msg.notify()方法唤醒处于阻塞状态的生产者线程。
写个测试代码,main方法:
- public class TestMain {
- public static void main(String[] args) {
- Queue<String>queue=new LinkedList<>();
- int maxSize=2;
- Product product=new Product(queue,maxSize);
- Consumer consumer=new Consumer(queue,maxSize);
-
- Thread t1=new Thread(product);
- Thread t2=new Thread(consumer);
- t1.start();
- t2.start();
- }
- }
生产者先生产数据,然后才能唤醒消费者线程,在生产了两个数据之后发现队列满了,此时通过wait()方法阻塞生产者线程。消费者线程开始运行,如果发现队列中的元素不为空,则进行数据的消费,同时唤醒生产者线程,如果消费者发现队列为空 ,则阻塞消费者线程。
生产者/消费者如果深入设计,可以非常复杂,例如,假如有多个消费者或者生产者该如何设计等等。
上面的生产者/消费者模型的执行过程是怎么样的呢?
我们这里用queue做为两者的共享队列,生产者和消费者都会对该共享队列进行读写操作。因此,为了保证原子性,生产者和消费者线程就必须要对该共享队列加锁,只有竞争到锁资源的线程才能加锁,也就是操作队列的数据。
假设生产者抢到了锁,则开始向队列添加数据,知道队列满了发生阻塞,也就是消费应该要消费数据了。在代码中生产者的wait()方法是在synchronized中调用的,此时锁是没有释放的,那么消费者获取不到同步锁该怎么办呢?其实在线程调用了wait()方法后,就会释放当前的同步锁。由于Consumer此时在synchronized同步队列中等待,所以Produer一旦释放锁就可以唤醒Consumer线程。Consumer被唤醒后需要去竞争锁资源,如果成功,则进行数据消费。
Consumer消费一个数据后会调用notify()方法,该方法只是唤醒处于阻塞状态下的线程,由于Consumer还没有释放锁,因此被唤醒的Producer需要等待Consumer释放锁之后才能继续沿着阻塞的位置开始执行。
过程图如下: