Java多线程：线程间通信方式

Posted 2022-12-04 杨 戬

文章目录

• Java线程通信
• • wait()、notify()、notifyAll()
  • • API说明
    • 实现原理
    • 代码实现
  • await()、signal()、signalAll()
  • • API说明
    • 实现原理
    • 代码实现
  • BlockingQueue
  • • API说明
    • 实现原理
    • 代码实现

Java线程通信

在Java中线程通信主要有以下三种方式：

wait()、notify()、notifyAll()

如果线程之间采用synchronized来保证线程安全，则可以利用wait()、notify()、notifyAll()来实现线程通信。

这三个方法都不是Thread类中所声明的方法，而是Object类中声明的方法。原因是每
个对象都拥有锁，所以让当前线程等待某个对象的锁，当然应该通过这个对象来操作，Object源码方法如下：

并且因为当前线程可能会等待多个线程的锁，如果通过线程来操作，就非常复杂了。另外，这三个方法都是本地方法，并且被final修饰，无法被重写。

API说明

方法	说明
wait()	方法可以让当前线程释放对象锁并进入阻塞状态
notify()	方法用于唤醒一个正在等待相应对象锁的线程，使其进入就绪队列，以便在当前线程释放锁后竞争锁，进而得到CPU的执行
notifyAll()	用于唤醒所有正在等待相应对象锁的线程，使它们进入就绪队列，以便在当前线程释放锁后竞争锁，进而得到CPU的执行

实现原理

每个锁对象都有两个队列，一个是就绪队列，一个是阻塞队列。就绪队列存储了已就绪（将要竞争锁）的线程，阻塞队列存储了被阻塞的线程。当一个阻塞线程被唤醒后，才会进入就绪队列，进而等待CPU的调度。反之，当一个线程被wait后，就会进入阻塞队列，等待被唤醒。

代码实现

案例步骤：通信是在不同线程间的通信，一个线程处于wait状态阻塞等待被唤醒，另一个线程通过notify或者notifyAll唤醒，当前的唤醒操作必须是作用与同一个对象，注意在进行唤醒和阻塞时必须要加锁的，加锁需要使用synchronized关键字。

代码如下：

package com.yyl.threadtest.communication;

class WaitDemo extends Thread
    private Object obj;

    public WaitDemo(Object obj) 
        this.obj = obj;
    

    @Override
    public void run() 
        synchronized (obj) 
            System.out.println(Thread.currentThread().getName() + "WaitDemo执行开始================");
            try 
                // 调用wait方法阻塞线程执行
                System.out.println(Thread.currentThread().getName() + "WaitDemo进行等待================");
                obj.wait();
             catch (InterruptedException e) 
                e.printStackTrace();
            
            System.out.println(Thread.currentThread().getName() + "WaitDemo执行结束================");
        
    


class NotifyDemo extends Thread
    private Object obj;

    public NotifyDemo(Object obj) 
        this.obj = obj;
    

    @Override
    public void run() 
        synchronized (obj) 
            System.out.println(Thread.currentThread().getName() + "NotifyDemo执行开始================");
            try 
                Thread.sleep(1000);
                System.out.println(Thread.currentThread().getName() + "NotifyDemo 睡眠1s后进行通知================");
             catch (InterruptedException e) 
                e.printStackTrace();
            
            obj.notify(); //调用notify方法唤醒阻塞线程
            System.out.println(Thread.currentThread().getName() + "NotifyDemo执行结束================");
        
    



public class WaitNotifyTest 
    public static void main(String[] args) 
        Object object = new Object();
        WaitDemo waitDemo = new WaitDemo(object);
        NotifyDemo notifyDemo = new NotifyDemo(object);
        waitDemo.setName("WaitDemo线程");
        notifyDemo.setName("NotifyDemo线程");

        waitDemo.start();
        notifyDemo.start();
    

运行结果：

await()、signal()、signalAll()

如果线程之间采用Lock来保证线程安全，则可以利用await()、signal()、signalAll()来实现线程通信。

这三个方法都是Condition接口中的方法，该接口是在Java 1.5中出现的，它用来替代传统的wait+notify实现线程间的协作，它的使用依赖于 Lock。相比使用wait+notify，使用Condition的await+signal这种方式能够更加安全和高效地实现线程间协作。

Condition依赖于Lock接口，生成一个Condition的基本代码是lock.newCondition() 。 必须要注意的是，Condition 的 await()/signal()/signalAll() 使用都必须在lock保护之内，也就是说，必须在lock.lock()和lock.unlock之间才可以使用。

事实上，await()/signal()/signalAll() 与 wait()/notify()/notifyAll()有着天然的对应关系。

• Conditon中的await()对应Object的wait()，
• Condition中的signal()对应Object的notify()
• Condition中的signalAll()对应Object的notifyAll()

所以他们的API是类似的：

API说明

方法	说明
await()	方法可以让当前线程释放lock锁并进入阻塞状态。
signal()	方法用于唤醒一个正在等待相应lock锁的线程，使其进入就绪队列，以便在当前线程释放锁后竞争锁，进而得到CPU的执行。
signalAll()	用于唤醒所有正在等待相应lock锁的线程，使它们进入就绪队列，以便在当前线程释放锁后竞争锁，进而得到CPU的执行。
lock.newCondition()()	用于获得Condition对象

实现原理

内容过多，参考我的另一篇博文：https://blog.csdn.net/weixin_45525272/article/details/127741915

代码实现

package com.yyl.threadtest.communication;

import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObject;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

class AwaitDemo extends Thread 
    private ReentrantLock lock;
    private Condition condition;

    public AwaitDemo(ReentrantLock lock, Condition condition) 
        this.lock = lock;
        this.condition = condition;
    

    @Override
    public void run() 
        lock.lock();
        System.out.println(Thread.currentThread().getName() + "AwaitDemo 执行开始================");
        try 
            // 调用wait方法阻塞线程执行
            System.out.println(Thread.currentThread().getName() + "AwaitDemo 进行等待================");
            condition.await();
         catch (InterruptedException e) 
            e.printStackTrace();
        
        System.out.println(Thread.currentThread().getName() + "AwaitDemo 执行结束================");
        lock.unlock();
    


class SignalDemo extends Thread 
    private ReentrantLock lock;
    private Condition condition;

    public SignalDemo(ReentrantLock lock, Condition condition) 
        this.lock = lock;
        this.condition = condition;
    

    @Override
    public void run() 
        lock.lock();
        System.out.println(Thread.currentThread().getName() + "SignalDemo 执行开始================");
        try 
            Thread.sleep(1000);
            System.out.println(Thread.currentThread().getName() + "SignalDemo 睡眠1s后进行通知================");

         catch (InterruptedException e) 
            e.printStackTrace();
        
        condition.signal(); //调用 signal 方法唤醒阻塞线程
        System.out.println(Thread.currentThread().getName() + "SignalDemo 执行结束================");
        lock.unlock();
    



public class AwaitSignalTest 
    public static void main(String[] args) 
        ReentrantLock lock = new ReentrantLock();
        Condition condition = lock.newCondition();
        AwaitDemo awaitDemo = new AwaitDemo(lock,condition);
        SignalDemo signalDemo = new SignalDemo(lock,condition);
        awaitDemo.setName("AwaitDemo线程");
        signalDemo.setName("SignalDemo线程");

        awaitDemo.start();
        signalDemo.start();
    

运行结果如下：

BlockingQueue

Java 5提供了一个BlockingQueue接口，虽然BlockingQueue也是Queue的子接口，但它的主要用途并不是作为容器，而是作为线程通信的工具。BlockingQueue具有一个特征：当生产者线程试图向BlockingQueue中放入元素时，如果该队列已满，则该线程被阻塞；当消费者线程试图从BlockingQueue中取出元素时，如果该队列已空，则该线程被阻塞。

程序的两个线程通过交替向BlockingQueue中放入元素、取出元素，即可很好地控制线程的通信。线程之间需要通信，最经典的场景就是生产者与消费者模型，而BlockingQueue就是针对该模型提供的解决方案。

API说明

核心方法：

方法	说明
put(anObject)	将参数放入队列,如果放不进去会阻塞
take()	取出第一个数据,取不到会阻塞

其他方法（与List类似）

方法\\处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除方法	remove()	poll()	take()	poll(time,unit)
检查方法	element()	peek()	不可用	不可用

多线程一般多用put与take

实现原理

内容过多，参考我的另一篇博文：https://blog.csdn.net/weixin_45525272/article/details/127741915

代码实现

class Demo02 
    public static void main(String[] args) throws Exception 
        // 创建阻塞队列的对象,容量为 1
        ArrayBlockingQueue<String> arrayBlockingQueue = new ArrayBlockingQueue<>(3);

        // 存储元素
        arrayBlockingQueue.put("糖果");

        // 取元素
        System.out.println(arrayBlockingQueue.take());
        System.out.println(arrayBlockingQueue.take()); // 取不到会阻塞

        System.out.println("程序结束了");
    

程序没有结束，他会一直读取阻塞队列