Java中控制多线程顺序执行
一、概述二、普通示例三、控制示例3.1、设置线程优先级3.2、使用线程类的join()3.2.1、在主线程join()3.2.2、在子线程join() 3.3、使用等待/通知的机制来阻塞线程3.3.1、使用synchronized+Object.wait()+Object.notify()3.3.2、ReetrantLock+Condition3.3.3、CountDownLatch3.3.4、Semaphore 3.4、使用单例线程池newSingleThreadExecutor 四、Cyclicbarrier一、概述
Java的线程调度策略是基于线程优先级的抢占式调度,同优先级的线程的执行顺序是看JVM心情的、不可预测的。那有什么办法来控制多线程按照开发者意愿的顺序执行呢?方法总比问题多:
首先设置线程的优先级肯定是一种选择,但最多只能控制10个不同优先级线程的顺序执行。Thread类中有提供一个插队的方法join(),那就可以利用这个插队的方法将线程排好执行顺序。 主线程join()子线程join() Java中有线程间的等待/通知的机制,那就做成一个等待/通知链来控制线程的执行顺序。 synchronized+Object.wait()+Object.notify()ReetrantLock+ConditionCountDownLatchSemaphore Java中有提供一个单例线程池newSingleThreadExecutor,它就是解决这个问题的优选。
在网上有些文章说Cyclicbarrier也可以实现控制顺序,但研究下来并不是同意义上的顺序,Cyclicbarrier是标记若干个目标点(目标点有需满足的线程数),在并发时,需等到规定线程数到达标记点,然后才可以继续往下执行。那谁先到达目标点还是要看JVM的心情啊,并且目标点后,所有线程继续执行,那执行顺序也是不受控的。(代码在最后有提供,可参考)
二、普通示例
这里先给一个不加任何控制操作的三个线程执行示例,方便对照测试:
public class ThreadSequence {volatile static Thread t0, t1, t2;public static void main(String[] args) {//初始化线程usePriority();//启动线程t0.start();t1.start();t2.start();}private static void usePriority() {t0 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName() + "执行完成");}});t1 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName() + "执行完成");}});t2 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName() + "执行完成");}});}}
可以运行多几次,就会发现会出现顺序错乱的现象。
三、控制示例
3.1、设置线程优先级
通过设置线程的优先级来控制执行顺序,但不能设置同一优先级,所以这个方案最多支持10个线程的顺序执行。优先级1-10从高到低顺序执行。【t0->t1->t2】
public class ThreadSequence {volatile static Thread t0,t1,t2;public static void main(String[] args) {usePriority();t0.setPriority(3);t1.setPriority(2);t2.setPriority(1);t1.start();t0.start();t2.start();}private static void usePriority(){t0 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"执行完成");}});t1 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"执行完成");}});t2 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"执行完成");}});}}
3.2、使用线程类的join()
Thread类中定义的join()方法,调用此方法使所属的线程对象进入执行run()方法中任务,而当前线程进入无限的阻塞,直到join()的线程执行完成。
3.2.1、在主线程join()
在主线程按顺序join()子线程,来控制线程按顺序执行:【t0->t1->t2】
public class ThreadSequence {volatile static Thread t0,t1,t2;public static void main(String[] args) {//使用joinuseJoin();t0.start();t0.join();t1.start();t1.join();t2.start();}//使用joinprivate static void useJoin(){t0 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"执行完成");}});t1 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"执行完成");}});t2 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"执行完成");}});}}
3.2.2、在子线程join()
在子线程前置子线程的join(),来控制线程按顺序执行:【t2->t1->t0】
public class ThreadSequence {volatile static Thread t0,t1,t2;public static void main(String[] args) {//使用joinuseJoin();t0.start();t1.start();t2.start();}//使用joinprivate static void useJoin(){t0 = new Thread(new Runnable() {@Overridepublic void run() {try {t1.join();} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"执行完成");}});t1 = new Thread(new Runnable() {@Overridepublic void run() {try {t2.join();} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"执行完成");}});t2 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"执行完成");}});}}
3.3、使用等待/通知的机制来阻塞线程
3.3.1、使用synchronized+Object.wait()+Object.notify()
wait()方法是Object类的方法,该方法用来将当前线程置入“预执行队列”中,并且在wait()所在的代码行处停止执行,直到接到通知或被中断为止。
在调用wait()之前,线程必须获取到该对象的对象级别锁,即只能在同步方法或同步块中调用wait()方法。在执行wait()方法后,当前线程释放锁。【t0->t1->t2】
public class ThreadSequence {volatile static Thread t0,t1,t2;static boolean t1Run = false,t2Run = false; //定义n-1个运行条件static Object lock1 = new Object(),lock2 = new Object(); //定义n-1个锁资源public static void main(String[] args) {useWait();t2.start();t0.start();t1.start();}private static void useWait(){t0 = new Thread(new Runnable() {@Overridepublic void run() {//t0执行synchronized (lock1){System.out.println(Thread.currentThread().getName()+"执行");//通知t1执行t1Run = true;//释放lock1lock1.notify();System.out.println(Thread.currentThread().getName()+"执行完成");}}});t1 = new Thread(new Runnable() {@Overridepublic void run() {synchronized (lock1){if (!t1Run){try {//lock1等待lock1.wait();System.out.println(Thread.currentThread().getName()+"进入等待");} catch (InterruptedException e) {e.printStackTrace();}}System.out.println(Thread.currentThread().getName()+"执行");synchronized (lock2) {//通知t2执行t2Run = true;//释放lock2lock2.notify();}System.out.println(Thread.currentThread().getName()+"执行完成");}}});t2 = new Thread(new Runnable() {@Overridepublic void run() {synchronized (lock2){if (!t2Run){//t2执行try {//lock2等待lock2.wait();System.out.println(Thread.currentThread().getName()+"进入等待");} catch (InterruptedException e) {e.printStackTrace();}}System.out.println(Thread.currentThread().getName()+"执行");System.out.println(Thread.currentThread().getName()+"执行完成");}}});}}
3.3.2、ReetrantLock+Condition
Condition可以实现多路通知,也就是在一个Lock对象里面可以创建多个Condition(即对象监视器)实例,线程对象可以注册在指定的Condition中,从而可以有选择性的进行线程通知,在调度线程上更加灵活。在使用notify/notifyAll通知时,被通知的线程却是由JVM随机选择的。【t0->t1->t2】
public class ThreadSequence {volatile static Thread t0,t1,t2;static boolean t1Run = false,t2Run = false; //定义n-1个运行条件static Lock lock = new ReentrantLock();static Condition condition1 = lock.newCondition();static Condition condition2 = lock.newCondition();public static void main(String[] args) {useCondition();t2.start();t0.start();t1.start();}private static void useCondition(){t0 = new Thread(new Runnable() {@Overridepublic void run() {//t0获取锁lock.lock();System.out.println(Thread.currentThread().getName()+"执行");//设置t1可运行t1Run = true;//通知t1运行condition1.signal();//t0释放锁lock.unlock();System.out.println(Thread.currentThread().getName()+"执行完成");}});t1 = new Thread(new Runnable() {@Overridepublic void run() {//t1获取锁lock.lock();try {//是否可运行状态if (!t1Run){//不可运行状态,进入等待condition1.await();System.out.println(Thread.currentThread().getName()+"进入等待");}System.out.println(Thread.currentThread().getName()+"执行");//设置t2为可运行t2Run = true;//通知t2运行condition2.signal();}catch (Exception e){e.printStackTrace();}finally {//t1释放锁lock.unlock();}System.out.println(Thread.currentThread().getName()+"执行完成");}});t2 = new Thread(new Runnable() {@Overridepublic void run() {//t2获取锁lock.lock();try {//是否可运行状态if (!t2Run){//不可运行状态,进入等待condition2.await();System.out.println(Thread.currentThread().getName()+"进入等待");}System.out.println(Thread.currentThread().getName()+"执行");}catch (Exception e){e.printStackTrace();}finally {//t2释放锁lock.unlock();}System.out.println(Thread.currentThread().getName()+"执行完成");}});}}
3.3.3、CountDownLatch
ountDownLatch 的作用是:当一个线程需要另外一个或多个线程完成后,再开始执行。比如主线程要等待一个子线程完成环境相关配置的加载工作,主线程才继续执行,就可以利用 CountDownLatch 来实现。
用到的方法:
CountDownLatch(int count):构造方法,创建一个值为count 的计数器await():阻塞当前线程,将当前线程加入阻塞队列。countDown():对计数器进行递减1操作,当计数器递减至0时,当前线程会去唤醒阻塞队列里的所有线程。
【t0->t1->t2】
public class ThreadSequence {volatile static Thread t0,t1,t2;/*** 计数器1 用于T0线程通知T1线程* 计数器2 用于T1线程通知T2线程* 注意:这里个数都设置成立1 ,当T0执行完成后,执行countDown,来通知T1线程*/static CountDownLatch countDownLatch1 = new CountDownLatch(1);static CountDownLatch countDownLatch2 = new CountDownLatch(1);public static void main(String[] args) {useCountDownLatch();t2.start();t1.start();t0.start();}private static void useCountDownLatch(){t0 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"执行");//唤醒阻塞线程t1countDownLatch1.countDown();System.out.println(Thread.currentThread().getName()+"执行完成");}});t1 = new Thread(new Runnable() {@Overridepublic void run() {try {countDownLatch1.await();System.out.println(Thread.currentThread().getName()+"进入等待");} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"执行");//唤醒阻塞线程t2countDownLatch2.countDown();System.out.println(Thread.currentThread().getName()+"执行完成");}});t2 = new Thread(new Runnable() {@Overridepublic void run() {try {//t2等待countDownLatch2.await();System.out.println(Thread.currentThread().getName()+"进入等待");} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"执行");System.out.println(Thread.currentThread().getName()+"执行完成");}});}}
3.3.4、Semaphore
Semaphore计数信号量,常用于限制可以访问某些资源(物理或逻辑的)线程数目。
用到的方法:
Semaphore(int permits):构造方法,permits就是允许同时运行的线程数目
public Semaphore(int permits,boolean fair):permits就是允许同时运行的线程数目 ,fair 是否为公平锁,如果是公平锁,那么获得锁的顺序与线程启动顺序有关
void acquire():从此信号量获取一个许可,在提供一个许可前一直将线程阻塞,否则线程被中断。
tryAcquire():尝试获得令牌,返回获取令牌成功或失败,不阻塞线程。
release() :释放一个令牌,唤醒一个获取令牌不成功的阻塞线程。
【t0->t1->t2】
public class ThreadSequence {volatile static Thread t0,t1,t2;static Semaphore semaphore1 = new Semaphore(0);static Semaphore semaphore2 = new Semaphore(0);public static void main(String[] args) {useSemaphore();t1.start();t0.start();t2.start();}private static void useSemaphore(){t0 = new Thread(new Runnable() {@Overridepublic void run() {try {//释放资源,semaphore1加1semaphore1.release();} catch (Exception e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"执行完成");}});t1 = new Thread(new Runnable() {@Overridepublic void run() {try {//获取资源,semaphore1减1,为0时进入阻塞semaphore1.acquire();} catch (InterruptedException e) {e.printStackTrace();}try {//释放资源,semaphore2加1semaphore2.release();} catch (Exception e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"执行完成");}});t2 = new Thread(new Runnable() {@Overridepublic void run() {try {//获取资源,semaphore2减1 为0时进入阻塞semaphore2.acquire();} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"执行完成");}});}}
3.4、使用单例线程池newSingleThreadExecutor
使用newSingleThreadExecutor线程池,由于核心线程数只有一个,所以只能按提交顺序执行。【t0->t1->t2】
public class ThreadSequence {volatile static Thread t0,t1,t2;//创建单例的线程池static ExecutorService executorService = Executors.newSingleThreadExecutor();public static void main(String[] args) {//按指定顺序提交useSingleThreadExecutor();executorService.submit(t0);executorService.submit(t1);executorService.submit(t2);//销毁executorService.shutdown();}//使用单例线程池-newSingleThreadExecutorprivate static void useSingleThreadExecutor(){t0 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"0执行完成");}});t1 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"1执行完成");}});t2 = new Thread(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+"2执行完成");}});}}
四、Cyclicbarrier
public class ThreadSequence {volatile static Thread t0,t1,t2;/*** 设置n-1个等待标志* 设置n-1个线程互相等待,直到到达同一个同步点,再继续一起执行。T0不执行完,T1就永远不会执行*/static CyclicBarrier cyclicBarrier1 = new CyclicBarrier(2);static CyclicBarrier cyclicBarrier2 = new CyclicBarrier(2);public static void main(String[] args) {useCyclicBarrier();t1.start();t0.start();t2.start();}//使用Cyclicbarrierprivate static void useCyclicBarrier(){t0 = new Thread(new Runnable() {@Overridepublic void run() {try {cyclicBarrier1.await();System.out.println(Thread.currentThread().getName()+"到达目的1,进入等待");} catch (Exception e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"执行");System.out.println(Thread.currentThread().getName()+"执行完成");}});t1 = new Thread(new Runnable() {@Overridepublic void run() {try {cyclicBarrier1.await();System.out.println(Thread.currentThread().getName() + "到达目的1,进入等待");System.out.println(Thread.currentThread().getName() + "执行");System.out.println(Thread.currentThread().getName() + "执行完成");cyclicBarrier2.await();System.out.println(Thread.currentThread().getName() + "到达目的2,进入等待");} catch (Exception e) {e.printStackTrace();}}});t2 = new Thread(new Runnable() {@Overridepublic void run() {try {cyclicBarrier2.await();System.out.println(Thread.currentThread().getName()+"到达目的2,进入等待");} catch (Exception e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"执行");System.out.println(Thread.currentThread().getName()+"执行完成");}});}}