java并发编程学习:如何等待多个线程执行完成后再继续后续处理(synchronized、join、FutureTask、CyclicBarrier)

多线程应用中,经常会遇到这种场景:后面的处理,依赖前面的N个线程的处理结果,必须等前面的线程执行完毕后,后面的代码才允许执行。

在我不知道CyclicBarrier之前,最容易想到的就是放置一个公用的static变量,假如有10个线程,每个线程处理完上去累加下结果,然后后面用一个死循环(或类似线程阻塞的方法),去数这个结果,达到10个,说明大家都爽完了,可以进行后续的事情了,这个想法虽然土鳖,但是基本上跟语言无关,几乎所有主流编程语言都支持。

package yjmyzz.test;

public class ThreadLockTest {

    public static int flag = 0;//公用变量

    public static void main(String[] args) throws Exception {
        ThreadLockTest testObj = new ThreadLockTest();
        final int threadNum = 10;

        for (int i = 0; i < threadNum; i++) {
            new Thread(new MyRunable(i, testObj)).start();
        }

        while (true) {
            if (testObj.flag >= threadNum) {
                System.out.println("-----------\n所有thread执行完成!");
                break;
            }
            Thread.sleep(10);
        }
    }

    static class MyRunable implements Runnable {
        int _i = 0;
        ThreadLockTest _test;

        public MyRunable(int i, ThreadLockTest test) {
            this._i = i;
            this._test = test;
        }

        @Override
        public void run() {
            try {
                Thread.sleep((long) (Math.random() * 10));
                System.out.println("thread " + _i + " done");
                //利用synchronized获得同步锁
                synchronized (_test) {
                    _test.flag += 1;
                }
                System.out.println("thread " + _i + " => " + _test.flag);//测试用
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

输出结果:

thread 0 done
thread 0 => 1
thread 9 done
thread 9 => 2
thread 1 done
thread 1 => 3
thread 3 done
thread 3 => 4
thread 7 done
thread 7 => 5
thread 6 done
thread 6 => 6
thread 2 done
thread 2 => 7
thread 4 done
thread 4 => 8
thread 8 done
thread 8 => 9
thread 5 done
thread 5 => 10
-----------
所有thread执行完成!

 

除了这个方法,还可以借助FutureTask,达到类似的效果,其get方法会阻塞线程,等到该异步处理完成。缺点就是,FutureTask调用的是Callable,必须要有返回值,所以就算你不想要返回值,也得返回点啥

package yjmyzz.test;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class FutureTaskTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        FutureTask<String>[] tasks = new FutureTask[10];

        for (int i = 0; i < tasks.length; i++) {
            final int j = i;
            tasks[i] = new FutureTask<String>(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    Thread.sleep((long) (Math.random() * 100));
                    return "task" + j + " done";
                }
            });
            new Thread(tasks[i]).start();
        }

        for (int i = 0; i < tasks.length; i++) {
            System.out.println(tasks[i].get());//依次等待所有task执行完毕
        }

        System.out.println("-----------\n所有task执行完成!");

    }
}

执行结果:

task0 done
task1 done
task2 done
task3 done
task4 done
task5 done
task6 done
task7 done
task8 done
task9 done
-----------
所有task执行完成!

  

此外,Thread的Join方法也可以实现类似的效果,主要代码如下:

    public static void main(String[] args) throws Exception {

        final int threadNum = 10;
        Thread[] threads = new Thread[threadNum];

        for (int i = 0; i < threadNum; i++) {
            threads[i] = new Thread(new MyRunable(i));
            threads[i].start();
        }

        for (int i = 0; i < threadNum; i++) {
            threads[i].join();
        }

        System.out.println("-----------\n所有thread执行完成!");

    }

 

当然,这个需求最“正统”的解法应该是使用CyclicBarrier,它可以设置一个所谓的“屏障点”(或称集合点),好比在一项团队活动中,每个人都是一个线程,但是规定某一项任务开始前,所有人必须先到达集合点,集合完成后,才能继续后面的任务。  

package yjmyzz.test;

import java.util.concurrent.CyclicBarrier;

public class ThreadTest {

    public static void main(String[] args) throws Exception {

        final int threadNum = 10;
        CyclicBarrier cb = new CyclicBarrier(threadNum + 1);//注意:10个子线程 + 1个主线程

        for (int i = 0; i < threadNum; i++) {
            new Thread(new MyRunable(cb, i)).start();
        }

        cb.await();
        System.out.println("-----------\n所有thread执行完成!");
    }

    static class MyRunable implements Runnable {
        CyclicBarrier _cb;
        int _i = 0;

        public MyRunable(CyclicBarrier cb, int i) {
            this._cb = cb;
            this._i = i;
        }

        @Override
        public void run() {
            try {
                Thread.sleep((long) (Math.random() * 100));
                System.out.println("thread " + _i + " done,正在等候其它线程完成...");
                _cb.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

thread 9 done,正在等候其它线程完成...
thread 5 done,正在等候其它线程完成...
thread 0 done,正在等候其它线程完成...
thread 6 done,正在等候其它线程完成...
thread 4 done,正在等候其它线程完成...
thread 2 done,正在等候其它线程完成...
thread 3 done,正在等候其它线程完成...
thread 8 done,正在等候其它线程完成...
thread 7 done,正在等候其它线程完成...
thread 1 done,正在等候其它线程完成...
-----------
所有thread执行完成!

 

参考文章:

http://ifeve.com/concurrency-cyclicbarrier/

http://ifeve.com/thread-synchronization-utilities-5/

http://ifeve.com/semaphore-countdownlatch-cyclicbarrier-phaser-exchanger-in-java/  

http://ifeve.com/thread-management-7/

时间: 2024-09-16 05:08:30

java并发编程学习:如何等待多个线程执行完成后再继续后续处理(synchronized、join、FutureTask、CyclicBarrier)的相关文章

Java 并发编程学习笔记之Synchronized简介_java

一.Synchronized的基本使用 Synchronized是Java中解决并发问题的一种最常用的方法,也是最简单的一种方法.Synchronized的作用主要有三个:(1)确保线程互斥的访问同步代码(2)保证共享变量的修改能够及时可见(3)有效解决重排序问题.从语法上讲,Synchronized总共有三种用法: (1)修饰普通方法 (2)修饰静态方法 (3)修饰代码块 接下来我就通过几个例子程序来说明一下这三种使用方式(为了便于比较,三段代码除了Synchronized的使用方式不同以外,

Java 并发编程学习笔记之核心理论基础_java

并发编程是Java程序员最重要的技能之一,也是最难掌握的一种技能.它要求编程者对计算机最底层的运作原理有深刻的理解,同时要求编程者逻辑清晰.思维缜密,这样才能写出高效.安全.可靠的多线程并发程序.本系列会从线程间协调的方式(wait.notify.notifyAll).Synchronized及Volatile的本质入手,详细解释JDK为我们提供的每种并发工具和底层实现机制.在此基础上,我们会进一步分析java.util.concurrent包的工具类,包括其使用方式.实现源码及其背后的原理.本

Java多线程--让主线程等待所有子线程执行完毕在执行_java

朋友让我帮忙写个程序从文本文档中导入数据到oracle数据库中,技术上没有什么难度,文档的格式都是固定的只要对应数据库中的字段解析就行了,关键在于性能. 数据量很大百万条记录,因此考虑到要用多线程并发执行,在写的过程中又遇到问题,我想统计所有子进程执行完毕总共的耗时,在第一个子进程创建前记录当前时间用System.currentTimeMillis()在最后一个子进程结束后记录当前时间,两次一减得到的时间差即为总共的用时,代码如下  long tStart = System.currentTim

Java多线程--让主线程等待所有子线程执行完毕代码

采用CountDownLatch类来实现     主线程    package test; import java.util.concurrent.CountDownLatch; public class Main {  /**   *   * @author Administrator/2012-3-1/上午09:36:55   */  public static void main(String[] args) {   int threadNum = 10;   CountDownLatch

java并发编程学习:用 Semaphore (信号量)控制并发资源

并发编程这方面以前关注得比较少,恶补一下,推荐一个好的网站:并发编程网 - ifeve.com,上面全是各种大牛原创或编译的并发编程文章. 今天先来学习Semaphore(信号量),字面上看,根本不知道这东西是干啥的,借用 并发工具类(三)控制并发线程数的Semaphore一文中的交通红绿信号灯的例子来理解一下: 一条4车道的主干道,假设100米长,每辆车假设占用的长度为10米(考虑到前后车距),也就是说这条道上满负载运行的话,最多只能容纳4*(100/10)=40辆车,如果有120辆车要通过的

java并发编程学习: ThreadLocal使用及原理

多线程应用中,如果希望一个变量隔离在某个线程内,即:该变量只能由某个线程本身可见,其它线程无法访问,那么ThreadLocal可以很方便的帮你做到这一点.  先来看一下示例: package yjmyzz.test; public class ThreadLocalTest1 { public static class MyRunnable implements Runnable { private ThreadLocal<Integer> threadLocal = new ThreadLo

JAVA并发编程学习笔记之CAS操作

CAS操作 CAS是单词compare and set的缩写,意思是指在set之前先比较该值有没有变化,只有在没变的情况下才对其赋值. 我们常常做这样的操作 if(a==b) { a++; } 试想一下如果在做a++之前a的值被改变了怎么办?a++还执行吗?出现该问题的原因是在多线程环境下,a的值处于一种不定的状态.采用锁可以解决此类问题,但CAS也可以解决,而且可以不加锁. int expect = a; if(a.compareAndSet(expect,a+1)) { doSomeThin

Java并发编程示例(四):可控的线程中断_java

在上一节"线程中断"中,我们讲解了如何中断一个正在执行的线程以及为了中断线程,我们必须对Thread动点什么手脚.一般情况下,我们可以使用上一节介绍的中断机制.但是,如果线程实现了一个分配到多个方法中的复杂算法,或者方法调用中有一个递归调用,我们应该使用更好的方式来控制线程的中断.为此,Java提供了InterruptedException异常.当检测到中断请求时,可以抛出此异常,并且在run()方法中捕获. 在本节,我们将使用一个线程查找指定目录及其子目录下文件来演示通过使用Inte

java并发编程学习: 守护线程(Daemon Thread)

在正式理解这个概念前,先把 守护线程 与 守护进程 这二个极其相似的说法区分开,守护进程通常是为了防止某些应用因各种意外原因退出,而在后台独立运行的系统服务或应用程序. 比如:我们开发了一个邮件发送程序,一直不停的监视队列池,发现有待发送的邮件,就将其发送出去.如果这个程序挂了(或被人误操作关了),邮件就不发出去了,为了防止这种情况,再开发一个类似windows 系统服务的应用,常驻后台,监制这个邮件发送程序是否在运行,如果没运行,则自动将其启动.   而我们今天说的java中的守护线程(Dae