线程间通信（也叫线程并发协作）的四种方式

2024-04-10 23:00:36

参考博客：https://crossoverjie.top/2018/03/16/java-senior/thread-communication/

线程间通信一共有以下四种方式：

1、wait()、notify() 、notifyAll() - 等待通知机制，这些方法属于java.lang.Object类的方法，只能在同步方法或者代码块中使用

2、Volatile 共享内存

3、CountDownLatch 并发工具

4、CyclicBarrier 并发工具

wait/notify

两个线程交替打印奇偶数,通过wait/notify实现

copypublic class WaitNotify {
  // 状态锁
  private static Object lock = new Object();
  private static Integer i = 0;

  public void odd() {
    while (i < 10) {
      synchronized (lock) {
        if (i % 2 == 1) {
          System.out.println(Thread.currentThread().getName() + " - " + i);
          try {
            Thread.sleep(1000);
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
          i++;
          lock.notify();
        } else {
          try {
            lock.wait();
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
      }
    }
  }

  public void even() {
    while (i < 10) {
      synchronized (lock) {
        if (i % 2 == 0) {
          System.out.println(Thread.currentThread().getName() + " - " + i);
          try {
            Thread.sleep(1000);
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
          i++;
          lock.notify();
        } else {
          try {
            lock.wait();
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
      }
    }
  }

  public static void main(String[] args) {

    WaitNotify waitNotify = new WaitNotify();

    Thread t1 = new Thread(() -> waitNotify.odd(), "线程1");
    Thread t2 = new Thread(() -> waitNotify.even(), "线程2");

    t1.start();
    t2.start();
  }
}

结果:

copy线程2 - 0
线程1 - 1
线程2 - 2
线程1 - 3
线程2 - 4
线程1 - 5
线程2 - 6
线程1 - 7
线程2 - 8
线程1 - 9
线程2 - 10

Volatile

Volatile 修饰内存可见性。因为 Java 是采用共享内存的方式进行线程通信的，所以可以采用以下方式用主线程关闭 A 线程:

copypublic class Volatile implements Runnable {
　　// flag采用volatile修饰，主要是为了内存可见性
  private static volatile Boolean flag = true;

  @Override
  public void run() {
    while (flag) {
      System.out.println(Thread.currentThread().getName() + " - 执行");
    }
    System.out.println("线程结束");
  }

  public static void main(String[] args) {
    Thread t = new Thread(new Volatile());
    t.start();
    try {
      Thread.sleep(5);
      flag = false;
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}
结果:

copyThread-0 - 执行
Thread-0 - 执行
Thread-0 - 执行
Thread-0 - 执行
Thread-0 - 执行
线程结束

CountDownLatch

CountDownLatch可以代替wait/notify的使用, 并去掉synchronized。 CountDownLatch 可以实现 join 相同的功能，但是更加的灵活。

CountDownLatch 也是基于 AQS(AbstractQueuedSynchronizer) 实现的

下面重写第一个例子:

copyimport java.util.concurrent.CountDownLatch;

public class CountDown {
  private static Integer i = 0;
  final static CountDownLatch countDown = new CountDownLatch(1);

  public void odd() {
    while (i < 10) {
      if (i % 2 == 1) {
        System.out.println(Thread.currentThread().getName() + " - " + i);
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        i++;
        countDown.countDown();
      } else {
        try {
          countDown.await();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }

  public void even() {
    while (i < 10) {
      if (i % 2 == 0) {
        System.out.println(Thread.currentThread().getName() + " - " + i);
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        i++;
        countDown.countDown();
      } else {
        try {
          countDown.await();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }

  public static void main(String[] args) {

    CountDown countDown = new CountDown();

    Thread t1 = new Thread(() -> countDown.odd(), "线程1");
    Thread t2 = new Thread(() -> countDown.even(), "线程2");

    t1.start();
    t2.start();
  }
}

CyclicBarrier 并发工具

CyclicBarrier 中文名叫做屏障或者是栅栏，也可以用于线程间通信。

它可以等待 N 个线程都达到某个状态后继续运行的效果。

首先初始化线程参与者。
调用 await() 将会在所有参与者线程都调用之前等待。
直到所有参与者都调用了 await() 后，所有线程从 await() 返回继续后续逻辑。

该工具可以实现 CountDownLatch 同样的功能，但是要更加灵活。甚至可以调用 reset() 方法重置 CyclicBarrier (需要自行捕获 BrokenBarrierException 处理) 然后重新执行。

copyimport java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest {
  public static void main(String[] args) {
    CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
    new Thread(() -> {
      System.out.println(Thread.currentThread().getName() + ": 准备...");
      try {
        cyclicBarrier.await();
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }
      System.out.println("全部启动完毕!");
    }, "线程1").start();

    new Thread(() -> {
      System.out.println(Thread.currentThread().getName() + ": 准备...");
      try {
        cyclicBarrier.await();
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }
      System.out.println("全部启动完毕!");
    }, "线程2").start();

    new Thread(() -> {
      System.out.println(Thread.currentThread().getName() + ": 准备...");
      try {
        cyclicBarrier.await();
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }
      System.out.println("全部启动完毕!");
    }, "线程3").start();
  }
}
结果:

copy线程3: 准备...
线程2: 准备...
线程1: 准备...
全部启动完毕!
全部启动完毕!
全部启动完毕!

码农公寓