java并发包工具之CountDownLatch/CyclicBarrier/Semaphore

几天前我在思考这样一个问题,如果用java的多线程来做一个合力完成的事,例如合并请求,假设前端来了3个请求,到java这一层需要合成一个请求并返回,那么需要启用3个线程去请求,并等所有的结果都返回了再合并一下返回给前端,这该如何去做?

最开始我找到的是Thread.join这个方法。

Thread.join这个方法在源码中的描述是

Waits for this thread to die.

意思是等这个线程执行完了再去执行主线程。
那么我们针对上述问题可以实现一个这样的版本

package thread;

/**
 * @author lkxiaolou
 */
public class ThreadPrint implements Runnable {

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " is running");
        try {
            Thread.sleep(5000L);
        } catch (Exception e) {
        }
    }

    public static void main(String[] args) {
        Thread thread1 = new Thread(new ThreadPrint());
        Thread thread2 = new Thread(new ThreadPrint());
        Thread thread3 = new Thread(new ThreadPrint());
        thread1.start();
        thread2.start();
        try {
            // 等待子线程完成后再执行主线程
            thread1.join();
            thread2.join();
        } catch (Exception e) {

        }
        thread3.start();
    }
}

先启动线程1和2,再等1、2执行完再启动第3个线程。
运行结果是这样的

Thread-0 is running
Thread-1 is running
Thread-1 done
Thread-0 done
Thread-2 is running
Thread-2 done

线程0和1不一定谁先执行完,但一定是在线程2执行前执行完成。

后来我又发现了java对多线程提供了一组并发包来做这些事情,其中有个叫CountDownLatch的类。这个类用起来简单也更加好理解,CountDownLatch提供了一个计数器,每当一个线程执行完计数器减1,直到减成0,主线程才可以开始执行。

package thread;

import java.util.concurrent.CountDownLatch;

/**
 * @author lkxiaolou
 */
public class ThreadPrintCountDownLatch implements Runnable {

    public static CountDownLatch countDownLatch = new CountDownLatch(2);

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " is running");
        try {
            Thread.sleep(1000L);
        } catch (Exception e) {

        }
        System.out.println(Thread.currentThread().getName() + " done");
        countDownLatch.countDown();
    }

    public static void main(String[] args) {
        Thread thread1 = new Thread(new ThreadPrintCountDownLatch());
        Thread thread2 = new Thread(new ThreadPrintCountDownLatch());
        Thread thread3 = new Thread(new ThreadPrintCountDownLatch());
        thread1.start();
        thread2.start();
        try {
            countDownLatch.await();
        } catch (Exception e) {

        }
        thread3.start();
    }
}

输出也和上面一样

Thread-0 is running
Thread-1 is running
Thread-0 done
Thread-1 done
Thread-2 is running
Thread-2 done

要注意减1的操作 countDownLatch.countDown(); 一定要放在关键部分最后,关键部分是你想等的操作已经完成了才行。countDownLatch有个缺点是计数器只能减不能加,一旦减到了0就不能再用了。

看了countDownLatch之后又发现了两个类似的类:CyclicBarrier和Semaphore。
CyclicBarrier,通过它可以实现让一组线程等待至某个状态之后再全部同时执行。

package thread;

import java.util.concurrent.CyclicBarrier;

/**
 * @author lkxiaolou
 */
public class ThreadPrintCyclicBarrier implements Runnable {

    public ThreadPrintCyclicBarrier(CyclicBarrier cyclicBarrier) {
        this.cyclicBarrier = cyclicBarrier;
    }

    private CyclicBarrier cyclicBarrier;

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " is running");
        try {
            Thread.sleep(1000L);
        } catch (Exception e) {

        }
        try {
            cyclicBarrier.await();
        } catch (Exception e) {

        }
        System.out.println(Thread.currentThread().getName() + " done");
    }

    public static void main(String[] args) {

        CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new ThreadPrint()); // 这个线程输出带括号

        Thread thread1 = new Thread(new ThreadPrintCyclicBarrier(cyclicBarrier));
        Thread thread2 = new Thread(new ThreadPrintCyclicBarrier(cyclicBarrier));
        thread1.start();
        thread2.start();
    }
}

输出是这样的

Thread-0 is running
Thread-1 is running
[Thread-0] is running
[Thread-0] done
Thread-0 done
Thread-1 done

可以看出CyclicBarrier可以让所有线程先执行,执行到cyclicBarrier.await再等待,等所有线程都执行到cyclicBarrier.await时执行我们指定的线程(也可以不指定),指定线程执行完了再把之前暂停的线程都起起来同时执行。似乎能做的事情更多。

最后我们看一下 Semaphore类,这个翻译成信号量,他能控制对临界资源的访问。

假若一个工厂有5台机器,但是有8个工人,一台机器同时只能被一个工人使用,只有使用完了,其他工人才能继续使用。那么我们就可以通过Semaphore来实现。

package thread;

import java.util.concurrent.Semaphore;

/**
 * @author lkxiaolou
 */
public class ThreadPrintSemaphore implements Runnable {

    public static Semaphore semaphore = new Semaphore(2);

    @Override
    public void run() {

        try {
            semaphore.acquire();
        } catch (Exception e) {

        }

        System.out.println(Thread.currentThread().getName() + " is running");
        try {
            Thread.sleep(1000L);
        } catch (Exception e) {

        }
        try {
            semaphore.release();
        } catch (Exception e) {

        }
        System.out.println(Thread.currentThread().getName() + " done");
    }

    public static void main(String[] args) {
        
        Thread thread1 = new Thread(new ThreadPrintSemaphore());
        Thread thread2 = new Thread(new ThreadPrintSemaphore());
        Thread thread3 = new Thread(new ThreadPrintSemaphore());
        thread1.start();
        thread2.start();
        thread3.start();
    }
}

输出结果:

Thread-0 is running
Thread-1 is running
Thread-1 done
Thread-2 is running
Thread-0 done
Thread-2 done

多次运行可以看出0和1线程必须有一个释放了,2才能开始运行。

©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

相关阅读更多精彩内容

友情链接更多精彩内容