等待某(N)个线程执行完再执行某个线程的几种方法(Thread.join(),CountDownLatch,CyclicBarrier,Semaphore)

时间:2023-03-09 07:04:35
等待某(N)个线程执行完再执行某个线程的几种方法(Thread.join(),CountDownLatch,CyclicBarrier,Semaphore)

1.main线程中先调用threadA.join() ,再调用threadB.join()实现A->B->main线程的执行顺序

调用threadA.join()时,main线程会挂起,等待threadA执行完毕返回后再执行,到执行threadB.join()时再挂起,待threadB执行完毕返回继续执行main

使用场景:线程B依赖线程A的计算结果的场景

package concurrency;

public class JoinTest {
public static void main(String[] args) throws InterruptedException{
Thread threadA = new Thread(new JoinJob(),"thread-A");
Thread threadB = new Thread(new JoinJob(),"thread-B");
threadB.start();
threadA.start();
threadA.join();
threadB.join(); System.out.println("main ending...");
}
} class JoinJob implements Runnable{ @Override
public void run() {
System.err.println(Thread.currentThread().getName() + " starting...");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " ending..."); } }

运行结果

等待某(N)个线程执行完再执行某个线程的几种方法(Thread.join(),CountDownLatch,CyclicBarrier,Semaphore)

2.CountDownLatch

使用场景:多线程数据计算时需要先分几个线程计算第一步骤的数据,再合并一个线程计算第二步

 package concurrency;

 import java.util.concurrent.CountDownLatch;

 public class CountDownLatchTest {
static CountDownLatch c = new CountDownLatch(6);//如果count大于实际线程数,c.await()会一直等待 public static void main(String[] args) throws InterruptedException{
for (int i = 0; i < c.getCount()/2; i++) {
new Thread(new CountDownJob(c)).start();
new Thread(new CountDownLatchJob2(c)).start(); } System.out.println("waiting " + c.getCount() + " subthread doing");
c.await();//await()会等待c的count值为0,才继续往下执行
System.out.println("all subthread done!");
System.out.println(Thread.currentThread().getName() + " done!");
}
} class CountDownJob implements Runnable{
CountDownLatch cDownLatch; public CountDownJob(CountDownLatch cDownLatch) {
super();
this.cDownLatch = cDownLatch;
} @Override
public void run() {
try {
System.out.println(Thread.currentThread().getName() + " is doing something...");
Thread.sleep(3000);
cDownLatch.countDown();//线程结束时cDownLatch-1
} catch (InterruptedException e) {
e.printStackTrace();
} } }
class CountDownLatchJob2 implements Runnable{
CountDownLatch cDownLatch; public CountDownLatchJob2(CountDownLatch cDownLatch) {
super();
this.cDownLatch = cDownLatch;
}
@Override
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
System.err.println("thread sleep exception");
}
System.out.println(Thread.currentThread().getName() + " is doing something");
cDownLatch.countDown();//线程结束时cDownLatch-1
} }

运行结果:

等待某(N)个线程执行完再执行某个线程的几种方法(Thread.join(),CountDownLatch,CyclicBarrier,Semaphore)

3.可循环使用的同步屏障CyclicBarrier

使用场景:多线程计算数据,等待全部线程执行完阶段一的工作之后,再执行后面的阶段工作

 package concurrency;

 import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier; public class CyclicBarrierTest {
public static void main(String[] args) throws InterruptedException, BrokenBarrierException{
CyclicBarrier cBarrier = new CyclicBarrier(6);
//CyclicBarrier 提供另一个构造函数传入所有线程执行完后需要执行的runnable
// CyclicBarrier cBarrier = new CyclicBarrier(6, new Runnable() {
//
// @Override
// public void run() {
// System.out.println("doing something after all threads done.");
// }
// });
for (int i = 0; i < cBarrier.getParties()/2; i++) {
new Thread(new CyclicBarrierJob1(cBarrier)).start();
new Thread(new CyclicBarrierJob2(cBarrier)).start();
}
//CyclicBarrier 可以重置再次使用,如计算发生错误时可以重置计数器,并让线程重新执行一次。
//而CountDownLatch只可以使用一次
//cBarrier.reset();
}
} class CyclicBarrierJob1 implements Runnable {
CyclicBarrier cBarrier; public CyclicBarrierJob1(CyclicBarrier cBarrier) {
super();
this.cBarrier = cBarrier;
} public void run() {
System.out.println(Thread.currentThread().getName() + " finished phase1");
try {
cBarrier.await();//执行完阶段1的工作之后,等待其他线程全部执行完阶段1,才会继续执行阶段二
} catch (InterruptedException e) {
System.err.println("interrupted!");
} catch (BrokenBarrierException e) {
System.err.println("broken barrier!");
} System.out.println(Thread.currentThread().getName() + " start phase2");
}
}
class CyclicBarrierJob2 implements Runnable {
CyclicBarrier cBarrier; public CyclicBarrierJob2(CyclicBarrier cBarrier) {
super();
this.cBarrier = cBarrier;
} public void run() {
System.out.println(Thread.currentThread().getName() + " finished phase1");
try {
cBarrier.await();//执行完阶段1的工作之后,等待其他线程全部执行完阶段1,才会继续执行阶段二
} catch (InterruptedException e) {
System.err.println("interrupted!");
} catch (BrokenBarrierException e) {
System.err.println("broken barrier!");
}
System.out.println(Thread.currentThread().getName() + " start phase2"); }
}

执行结果:

等待某(N)个线程执行完再执行某个线程的几种方法(Thread.join(),CountDownLatch,CyclicBarrier,Semaphore)

4.可控制并发线程数的信号量Semaphore

使用场景:控制同时访问特定资源的线程数量

 package concurrency;

 import java.util.concurrent.Semaphore;

 public class SemaphoreTest {
public static void main(String[] args) {
final Semaphore semaphore = new Semaphore(3);//限制可以同时访问资源的线程数是10个 for (int i = 0; i < 12; i++) {//新建30个线程
new Thread(new Runnable() { @Override
public void run() {
try {
semaphore.acquire();//获取到许可证才可以访问资源
System.out.println("accessing the database ..");
Thread.sleep(3000);
semaphore.release();//资源访问结束之后释放许可证
} catch (InterruptedException e) {
System.out.println("interrupted!");
} }
}).start();
} }
}

等待某(N)个线程执行完再执行某个线程的几种方法(Thread.join(),CountDownLatch,CyclicBarrier,Semaphore)