架构系列——线程通信的实现方式

2023-08-30 55

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 架构系列——线程通信的实现方式

一、什么是线程通信？

线程是操作系统中独立的个体，但这些个体如果不经过特殊的处理就不能称为一个整体，线程间的通信就成为整体的必用方式之一。

当线程存在通信指挥，系统间的交互性会更强大，在提高CPU利用率的同时，还会使开发人员在处理线程任务的过程中有效的把控和监督。

二、线程通信的实现

1.使用volatile实现线程通信

如下列代码所示，list使用volatile修饰，利用volatile的可见性，当线程t1添加了5个元素的时候，t2得到通知！

弊端：需要不断地做while循环，影响性能！

public class ThreadCommunication {
  private volatile static List list = new ArrayList();
  public static void main(String[] args) {
    final ThreadCommunication ThreadCommunication = new ThreadCommunication();
    Thread t1 = new Thread(new Runnable(){
      @Override
      public void run() {
        // TODO Auto-generated method stub
        try {
          for (int i = 0; i < 10; i++) {
            list.add("test");
            System.out.println("当前线程" + Thread.currentThread().getName() + "添加了一个元素");
            Thread.sleep(500);
          }
        } catch (InterruptedException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
      }
    }, "t1");
    Thread t2 = new Thread(new Runnable(){
      @Override
      public void run() {
        // TODO Auto-generated method stub
        while(true){
          if(list.size() == 5){
            System.out.println("当前线程" + Thread.currentThread().getName() + "收到通知，list数量为5");
            throw new RuntimeException();
          }
        }
      }
    }, "t2");
    t2.start();
    t1.start();
  }
}

2.使用wait/notify实现线程通信

wait和notify都是Object类的方法，而Object是所有类的父类，所以，java为所有的对象提供了这两个方法。

1、wait和notify必须配合synchronized使用
2、wait方法释放锁，notify方法不释放锁

如下列代码所示，使用wait、notify结合synchronized实现线程通信。

弊端：不能实时获得锁，下面的代码运行之后，t1的循环做完t2才得到通知。

public class ThreadCommunication2 {
  private static List list = new ArrayList();
  public static void main(String[] args) {
    final Object lock = new Object();
    Thread t1 = new Thread(new Runnable(){
      @Override
      public void run() {
        // TODO Auto-generated method stub
        try {
          synchronized(lock){
            for (int i = 0; i < 10; i++) {
              list.add("test");
              System.out.println("当前线程" + Thread.currentThread().getName() + "添加了一个元素");
              Thread.sleep(500);
              if(list.size() == 5){
                System.out.println("已经发出通知");
                lock.notify();
              }
            }
          }
        } catch (InterruptedException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
      }
    }, "t1");
    Thread t2 = new Thread(new Runnable(){
      @Override
      public void run() {
        // TODO Auto-generated method stub
        synchronized(lock){
          if(list.size() != 5){
            try {
              lock.wait();
            } catch (InterruptedException e) {
              // TODO Auto-generated catch block
              e.printStackTrace();
            }
          }
          System.out.println("当前线程" + Thread.currentThread().getName() + "收到通知，list数量为5");
          throw new RuntimeException();
        }
      }
    }, "t2");
    t2.start();
    t1.start();
  }
}

3.使用ReentrantLock创建Condition实现

Conditon中的await()对应Object的wait()；

Condition中的signal()对应Object的notify()；

Condition中的signalAll()对应Object的notifyAll()。

这种方法跟 Object 的 wait和 notify 一样，不能实时获得锁，下面的代码运行之后，t1的循环做完t2才得到通知。

public class ThreadCommunication3 {
    public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();
        Condition condition = lock.newCondition();
        List<String> list = new ArrayList<>();
        Thread t1 = new Thread(() -> {
            lock.lock();
            for (int i = 1; i <= 10; i++) {
                list.add("abc");
                System.out.println("线程1向list中添加一个元素，此时list中的元素个数为：" + list.size());
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (list.size() == 5){
                    condition.signal();
                }
            }
            lock.unlock();
        });
        // 实现线程B
        Thread t2 = new Thread(() -> {
            lock.lock();
            if (list.size() != 5) {
                try {
                    condition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("线程2收到通知...");
            lock.unlock();
        });
        t2.start();
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        t1.start();
    }
}

4.通过LockSupport实现线程通信

LockSupport是一个线程阻塞工具类，所有的方法都是静态方法，可以让线程在任意位置阻塞，当然阻塞之后肯定得有唤醒的方法。

线程通信主要用到两个函数：

park：英文意思为停车。我们如果把Thread看成一辆车的话，park就是让车停下

unpark：就是让车启动然后跑起来

弊端：需要知道被唤起的线程名字

public class ThreadCommunication4 {
  public static void main(String[] args) {
    List<String> list = new ArrayList<>();
    final Thread t2 = new Thread(() -> {
      if (list.size() != 5) {
        //线程阻塞
        LockSupport.park();
      }
      System.out.println("线程2收到通知...");
    });
    Thread t1 = new Thread(() -> {
      for (int i = 1; i <= 10; i++) {
        list.add("abc");
        System.out.println("线程1向list中添加一个元素，此时list中的元素个数为：" + list.size());
        try {
          Thread.sleep(100);
        } catch (Exception e) {
          e.printStackTrace();
        }
        if (list.size() == 5){
          //唤起线程t2
          LockSupport.unpark(t2);
        }
      }
    });
    t1.start();
    t2.start();
  }
}

5.使用CountDownLatch实现线程通信

CountDownLatch是concurrent包里面的一个类，CountDownLatch能够使一个线程在等待另外一些线程完成各自工作之后，再继续执行。当每一个线程完成自己任务后，计数器的值就会减1。当计数器的值为0时，表示所有的线程都已经完成一些任务，然后在CountDownLatch上等待的线程就可以唤醒其他线程，继续执行接下来的任务。

await()函数：此函数将会使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断。

countDown函数：此函数将递减锁存器的计数，如果计数到达零，则释放所有等待的线程

如下列代码所示，t1中使用countDown方法之后，计数器会减到0，这时会立即唤醒t2线程，然后继续执行t2线程的代码

public class ThreadCommunication5 {
  private static List list = new ArrayList();
  public static void main(String[] args) {
    final CountDownLatch cdl = new CountDownLatch(1);//设置一个计数器
    Thread t1 = new Thread(new Runnable(){
      @Override
      public void run() {
        // TODO Auto-generated method stub
        try {
          for (int i = 0; i < 10; i++) {
            list.add("test");
            System.out.println("当前线程" + Thread.currentThread().getName() + "添加了一个元素");
            Thread.sleep(500);
            if(list.size() == 5){
              System.out.println("已经发出通知");
              cdl.countDown();//计数器变为0，立即唤醒t2线程             
            }
          }
        } catch (InterruptedException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
      }
    }, "t1");
    Thread t2 = new Thread(new Runnable(){
      @Override
      public void run() {
        // TODO Auto-generated method stub
        if(list.size() != 5){
          try {
            cdl.await();//在计数器等于0之前，会一直处于等待状态 
          } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
          }
        }
        System.out.println("当前线程" + Thread.currentThread().getName() + "收到通知，list数量为5");
        throw new RuntimeException();
      }
    }, "t2");
    t2.start();
    t1.start();
  }
}