Java并发编程系列- 线程的并发工具类

Posted way2backend

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Java并发编程系列- 线程的并发工具类相关的知识,希望对你有一定的参考价值。

技术图片

2.线程的并发工具类

2.1 Fork-Join

JDK 7中引入了fork-join框架,专门来解决计算密集型的任务。可以将一个大任务,拆分成若干个小任务,如下图所示:

技术图片

Fork-Join框架利用了分而治之的思想:什么是分而治之?规模为N的问题,N<阈值,直接解决,N>阈值,将N分解为K个小规模子问题,子问题互相对立,与原问题形式相同,将子问题的解合并得到原问题的解.

具体使用中,需要向ForkJoinPool线程池提交一个ForkJoinTask任务。ForkJoinTask任务有两个重要的子类,RecursiveAction类和RecursiveTask类,他们分别表示没有返回值的任务和可以有返回值的任务。

RecursiveAction类

下面的例子中,我们使用RecursiveAction遍历指定目录来寻找特定类型文件,需要实现compute方法。

public class FindDirsFiles extends RecursiveAction{

    private File path;//当前任务需要搜寻的目录

    public FindDirsFiles(File path) {
        this.path = path;
    }

    public static void main(String [] args){
        try {
            // 用一个 ForkJoinPool 实例调度总任务
            ForkJoinPool pool = new ForkJoinPool();
            FindDirsFiles task = new FindDirsFiles(new File("F:/"));

            pool.execute(task);//异步调用

            System.out.println("Task is Running......");
            Thread.sleep(1);
            int otherWork = 0;
            for(int i=0;i<100;i++){
                otherWork = otherWork+i;
            }
            System.out.println("Main Thread done sth......,otherWork="+otherWork);
            task.join();//阻塞的方法
            System.out.println("Task end");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    protected void compute() {
        
        List<FindDirsFiles> subTasks = new ArrayList<>();
        
        File[] files = path.listFiles();
        if(files!=null) {
            for(File file:files) {
                if(file.isDirectory()) {
                    subTasks.add(new FindDirsFiles(file));
                }else {
                    //遇到文件,检查
                    if(file.getAbsolutePath().endsWith("txt")) {
                        System.out.println("文件:"+file.getAbsolutePath());
                    }
                }
            }
            if(!subTasks.isEmpty()) {
                for(FindDirsFiles subTask:invokeAll(subTasks)) {
                    subTask.join();//等待子任务执行完成
                }
            }
        }
    }
}

RecursiveTask类

下面的例子中,利用RecursiveTask来实现数值累加。

public class SumArray {
    private static class SumTask extends RecursiveTask<Integer>{

        private final static int THRESHOLD = MakeArray.ARRAY_LENGTH/10;
        private int[] src; //表示我们要实际统计的数组
        private int fromIndex;//开始统计的下标
        private int toIndex;//统计到哪里结束的下标

        public SumTask(int[] src, int fromIndex, int toIndex) {
            this.src = src;
            this.fromIndex = fromIndex;
            this.toIndex = toIndex;
        }

        @Override
        protected Integer compute() {
            if(toIndex-fromIndex < THRESHOLD) {
                int count = 0;
                for(int i=fromIndex;i<=toIndex;i++) {
                    //SleepTools.ms(1);
                    count = count + src[i];
                }
                return count;
            }else {
                //fromIndex....mid....toIndex
                //1...................70....100
                int mid = (fromIndex+toIndex)/2;
                SumTask left = new SumTask(src,fromIndex,mid);
                SumTask right = new SumTask(src,mid+1,toIndex);
                invokeAll(left,right);
//              left.fork();
//              right.fork();
                return left.join()+right.join();
            }
        }
    }


    public static void main(String[] args) {

        ForkJoinPool pool = new ForkJoinPool();
        int[] src = MakeArray.makeArray();

        SumTask innerFind = new SumTask(src,0,src.length-1);

        long start = System.currentTimeMillis();

        pool.invoke(innerFind);//同步调用
        System.out.println("Task is Running.....");

        System.out.println("The count is "+innerFind.join()
                +" spend time:"+(System.currentTimeMillis()-start)+"ms");

    }
}

注意到fork和invokeAll都能达到相同的效果,只是fork将task交给工作线程后立刻返回;但是invokeAll会fork其中一个任务后,同时同步的调用另外一个任务,然后等待两个任务完成,可以参考invokeAll的实现:

ublic static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
        int s1, s2;
        t2.fork();
        if (((s1 = t1.doInvoke()) & ABNORMAL) != 0)
            t1.reportException(s1);
        if (((s2 = t2.doJoin()) & ABNORMAL) != 0)
            t2.reportException(s2);
    }

工作密取(Work Stealing)

在后台,fork-join框架使用了一种有效的方法来平衡可用线程的负载,称为工作密取(Work stealing)。每个工作线程都有一个双端队列(deque)来完成任务,一个工作线程将子任务压入其双端队列的队头。当一个工作线程空闲时,它会从另一个双端队列的队尾密取一个任务。

ForkJoinPool内部利用循环数组实现了一个双端队列,称为WorkQueue。对于这个Queue,有3种操作方法,分别是push、pop和poll。对于push和pop操作,只能被拥有该Queue的线程所调用。poll操作被用于其他工作线程从该Queue中获得task。

考虑到多线程steal work的情况,当进行poll操作时,会通过CAS操作来保证多线程下的安全性。如果CAS操作成功,则说明窃取成功。

技术图片

更多细节可以查看ForkJoinPool的实现以及论文https://www.dre.vanderbilt.edu/~schmidt/PDF/work-stealing-dequeue.pdf。

invoke && execute && submit 区别

invoke是同步调用,它会马上执行执行,并且将task join到当前线程,也就是阻塞当前线程。

    /**
     * Performs the given task, returning its result upon completion.
     * If the computation encounters an unchecked Exception or Error,
     * it is rethrown as the outcome of this invocation.  Rethrown
     * exceptions behave in the same way as regular exceptions, but,
     * when possible, contain stack traces (as displayed for example
     * using {@code ex.printStackTrace()}) of both the current thread
     * as well as the thread actually encountering the exception;
     * minimally only the latter.
     *
     * @param task the task
     * @param <T> the type of the task's result
     * @return the task's result
     * @throws NullPointerException if the task is null
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     */
    public <T> T invoke(ForkJoinTask<T> task) {
        if (task == null)
            throw new NullPointerException();
        externalSubmit(task);
        return task.join();
    }

execute和submit是异步调用,它会将Task送到Work Queue中等待运行。如果需要看到运行结果,可以在execute和submit后调用join方法。两者的区别只是submit会返回task,execute返回空值。

    /**
     * Submits a ForkJoinTask for execution.
     *
     * @param task the task to submit
     * @param <T> the type of the task's result
     * @return the task
     * @throws NullPointerException if the task is null
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     */
    public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
        return externalSubmit(task);
    }

下面是execute的实现:

    /**
     * Arranges for (asynchronous) execution of the given task.
     *
     * @param task the task
     * @throws NullPointerException if the task is null
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     */
    public void execute(ForkJoinTask<?> task) {
        externalSubmit(task);
    }

2.2 CountDownLatch

Latch是门栓的意思,顾名思义,CountDownLatch是一个多线程的控制工具类。通常用于让一组线程等待直到倒计时结束,再开始执行。

技术图片

CountDownLatch的用法如下,

  1. 初始化count down的次数;
  2. 在初始化线程中调用countDown对计数器进行减1;
  3. 工作线程中调用await进行等待,当计时器为0时,工作线程开始工作。
public class UseCountDownLatch {
    
    static CountDownLatch latch = new CountDownLatch(6);

    // 初始化线程(只有一步,有4个)
    private static class InitThread implements Runnable{

        @Override
        public void run() {
            System.out.println("Thread_"+Thread.currentThread().getId() +" finish init work......");
            
            latch.countDown();//初始化线程完成工作了,countDown方法只扣减一次;
            
            // We can add some tasks after the countDown is invoked
            for(int i =0;i<2;i++) {
                System.out.println("Thread_"+Thread.currentThread().getId() +" ........continue do its work");
            }
        }
    }
    
    // 业务线程
    private static class BusinessThread implements Runnable{
        @Override
        public void run() {
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            
            for(int i =0;i<3;i++) {
                System.out.println("BusinessThread_"+Thread.currentThread().getId() +" start to do business-----");
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 单独的初始化线程,初始化分为2步,需要扣减两次
        new Thread(new Runnable() {
            @Override
            public void run() {
                SleepTools.ms(1);
                System.out.println("Thread_"+Thread.currentThread().getId() +" finish init work step 1st......");
                latch.countDown();//每完成一步初始化工作,扣减一次
                
                System.out.println("begin step 2nd.......");
                SleepTools.ms(1);
                System.out.println("Thread_"+Thread.currentThread().getId() +" finish init work step 2nd......");
                latch.countDown();//每完成一步初始化工作,扣减一次
            }
        }).start();
        
        new Thread(new BusinessThread()).start();
        
        // Start 3 new init thread
        for(int i=0;i<=3;i++){
            Thread thread = new Thread(new InitThread());
            thread.start();
        }

        latch.await();
        
        System.out.println("Main do ites work........");
    }
}

2.3 CyclicBarrier

CyclicBarrier类实现了一个集结点,称为屏障(barrier)。当一个线程完成了那部分任务之后,它运行到屏障处,一旦所有线程都到达了这个屏障,屏障就撤销,线程就可以继续运行了。

技术图片

CyclicBarrier的用法如下:

  1. 构造一个Barrier,需要给出参与的线程数。JDK里提供了两个构造函数,barrierAction为屏障打开之后需要执行的action。
CyclicBarrier(int parties) 

CyclicBarrier(int parties, Runnable barrierAction)
  1. 每个线程做一些事情,完成后在屏障上调用await等待,
public void run() {
    doSomeWork();
    barrier.await();
    ...
}
  1. 当所有线程都到达了await后,此时屏障打开。如果有定义屏障打开后执行的action,则会先执行action。然后其他线程继续往下执行await后面的部分。

下面是具体的例子:

在打开屏障后,输出了各个线程的id。

public class UseCyclicBarrier {
    
    private static CyclicBarrier barrier = new CyclicBarrier(5,new TaskAfterBarrierIsOpenThread());
    
    private static ConcurrentHashMap<String,Long> resultMap = new ConcurrentHashMap<>();//存放子线程工作结果的容器

    public static void main(String[] args) {
        for(int i=0;i< 5;i++){
            Thread thread = new Thread(new SubThread());
            thread.start();
        }
    }

    //负责屏障开放以后的工作
    private static class TaskAfterBarrierIsOpenThread implements Runnable{

        @Override
        public void run() {
            StringBuilder result = new StringBuilder();
            for(Map.Entry<String,Long> workResult:resultMap.entrySet()){
                result.append("["+workResult.getValue()+"]");
            }
            System.out.println(" the result = "+ result);
            System.out.println("do other business........");
        }
    }

    //工作线程
    private static class SubThread implements Runnable{

        @Override
        public void run() {
            long id = Thread.currentThread().getId();//线程本身的处理结果
            resultMap.put(Thread.currentThread().getId()+"",id);
            Random r = new Random();//随机决定工作线程的是否睡眠
            try {
                if(r.nextBoolean()) {
                    Thread.sleep(2000+id);
                    System.out.println("Thread_"+id+" ....do something ");
                }
                System.out.println(id+"....is await");
                barrier.await();
                Thread.sleep(1000+id);
                System.out.println("Thread_"+id+" ....do its business ");
            } catch (Exception e) {
                e.printStackTrace();
            }

        }
    }
}

注意屏障是可以重复使用的,当所有等待线程被释放后可以被重用。

CountDownLatch和CyclicBarrier对比
countdownlatch的放行由第三者控制,CyclicBarrier放行由一组线程本身控制
countdownlatch放行条件 >= 线程数,CyclicBarrier放行条件 = 线程数

2.4 Semaphore

Semaphore也叫信号量,在JDK1.5被引入,可以用来控制同时访问特定资源的线程数量,通过协调各个线程,以保证合理的使用资源。

Semaphore内部维护了一组虚拟的许可,许可的数量可以通过构造函数的参数指定。

  • 访问特定资源前,必须使用acquire方法获得许可,如果许可数量为0,该线程则一直阻塞,直到有可用许可。
  • 访问资源后,使用release释放许可。

示例程序如下:

public class MySemaphoreTest {
    static Semaphore semaphore = new Semaphore(4);
    
    private static class BusinessThread extends Thread {
        String name = "";

        BusinessThread(String name) {
            this.name = name;
        }
        
        @Override
        public void run() {
            try {
                System.out.println(name + " try to acquire lock...");
                System.out.println(name + " : available Semaphore permits now: " + semaphore.availablePermits());
                
                semaphore.acquire();
                
                System.out.println(name + " : got the permit!");            
                
                // Do some business work
                Thread.sleep(1000);
                
                System.out.println(name + " : release lock...");
                semaphore.release();
                
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    
    public static void main(String[] args) {
        for (int i = 0; i < 6; i++) {
            new BusinessThread((i+1) + "").start();
        }
    }
}

2.5 Exchanger

当两个线程在同一个数据缓冲区的两个实例上工作时,就可以使用Exchanger。典型的情况是,一个线程向缓冲区填入数据,另一个线程消耗这些数据。当他们都完成之后,相互交换缓冲区。

下面的例子中,在两个线程中分别填入数据,然后交换数据,最后打印从对方线程交换得来的数据。

public class UseExchange {
    private static final Exchanger<Set<String>> exchange 
        = new Exchanger<Set<String>>();

    public static void main(String[] args) {

        //第一个线程
        new Thread(new Runnable() {
            @Override
            public void run() {
                Set<String> setA = new HashSet<String>();//存放数据的容器
                try {
                    setA.add("1");
                    System.out.println(Thread.currentThread().getName()  + " Add 1 to the set");                    
                    setA.add("2");
                    System.out.println(Thread.currentThread().getName()  + " Add 2 to the set");
                    setA.add("3");
                    System.out.println(Thread.currentThread().getName()  + " Add 3 to the set");
                       
                    setA = exchange.exchange(setA);//交换setA出去,返回交换来的数据setB
                    
                    /*处理交换后的数据*/
                    System.out.println(Thread.currentThread().getName()  + " print the data after exchange ");
                    setA.forEach(string -> System.out.println(Thread.currentThread().getName() + " print" + string));
                } catch (InterruptedException e) {
                }
            }
        }).start();

      //第二个线程
        new Thread(new Runnable() {
            @Override
            public void run() {
                Set<String> setB = new HashSet<String>();//存放数据的容器
                try {
                    setB.add("A");
                    System.out.println(Thread.currentThread().getName()  + " Add A to the set");
                    
                    setB.add("B");
                    System.out.println(Thread.currentThread().getName()  + " Add B to the set");
                    
                    setB.add("C");
                    System.out.println(Thread.currentThread().getName()  + " Add C to the set");
                    
                    setB = exchange.exchange(setB);//交换setB出去,返回交换来的数据setA
                    
                    /*处理交换后的数据*/
                    System.out.println(Thread.currentThread().getName()  + " print the data after exchange ");
                    setB.forEach(string -> System.out.println(Thread.currentThread().getName() + " print" + string));
                } catch (InterruptedException e) {
                }
            }
        }).start();
    }
}

2.6 Future、Callable和FutureTask

Future是多线程开发中的一种常见设计模式,核心思想是异步调用。当需要调用一个函数方法时,如果这个函数很慢,需要进行等待,这时可以先处理一些其他任务,在真正需要数据的时候再去尝试获得需要的数据。

技术图片

以上是Future的基本结构,RunnableFuture继承了Future和Runnable接口,FutureTask可以接收一个Callable实例作为运行的任务。

Future的使用比较简单,例子如下:

public class UseFuture {
    
    /*实现Callable接口,允许有返回值*/
    private static class UseCallable implements Callable<Integer>{

        private int sum;
        @Override
        public Integer call() throws Exception {
            System.out.println("Callable子线程开始计算");
            Thread.sleep(2000);
            for(int i=0;i<5000;i++) {
                sum = sum+i;
            }
            System.out.println("Callable子线程计算完成,结果="+sum);
            return sum;
        }

    }
    
    public static void main(String[] args) 
            throws InterruptedException, ExecutionException {
        
        UseCallable useCallable = new UseCallable();
        FutureTask<Integer> futureTask = new FutureTask<Integer>(useCallable);
        new Thread(futureTask).start();
        
        Random r = new Random();
        SleepTools.second(1);
        if(r.nextBoolean()) {//随机决定是获得结果还是终止任务
            System.out.println("Get UseCallable result = "+futureTask.get());
        }else {
            System.out.println("中断计算");
            futureTask.cancel(true);
        }
    }
}

注意Future接口中声明了5个方法,分别为:

  • cancel方法:用来取消任务,如果取消任务成功则返回true,如果取消任务失败则返回false。参数mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务,如果设置true,则表示可以取消正在执行过程中的任务。如果任务已经完成,则无论mayInterruptIfRunning为true还是false,此方法肯定返回false,即如果取消已经完成的任务会返回false;如果任务正在执行,若mayInterruptIfRunning设置为true,则返回true,若mayInterruptIfRunning设置为false,则返回false;如果任务还没有执行,则无论mayInterruptIfRunning为true还是false,肯定返回true。
  • isCancelled方法:表示任务是否被取消成功,如果在任务正常完成前被取消成功,则返回 true。
  • isDone方法:表示任务是否已经完成,若任务完成,则返回true;
  • get()方法:用来获取执行结果,这个方法会产生阻塞,会一直等到任务执行完毕才返回;
  • get(long timeout, TimeUnit unit):用来获取执行结果,如果在指定时间内,还没获取到结果,就直接返回null。

本文由『后端精进之路』原创,首发于博客 http://teckee.github.io/ , 转载请注明出处

搜索『后端精进之路』关注公众号,立刻获取最新文章和价值2000元的BATJ精品面试课程

技术图片

以上是关于Java并发编程系列- 线程的并发工具类的主要内容,如果未能解决你的问题,请参考以下文章

Java多线程系列:CountDownLatchSemaphore等4大并发工具类详解

『死磕Java并发编程系列』并发编程工具类之CountDownLatch

『死磕Java并发编程系列』并发编程工具类之CountDownLatch

『死磕Java并发编程系列』并发编程工具类之CountDownLatch

『死磕Java并发编程系列』并发编程工具类之CountDownLatch

『死磕Java并发编程系列』并发编程工具类之CountDownLatch