[多线程] 多线程工具之CountDownLatch

心已赠人 2022-09-15 11:44 289阅读 0赞

### 前言 ###

在Java中, 如何启动多个线程. 随后使某项任务分步执行.

(TODO 此处可以来2个说明图和小故事.)

--------------------

### 使用Join方法 ###

public static Main(String []args){
    	// 声明线程
    	Thread threadA1 = new Thread();
    	Thread threadA2 = new Thread();
    	Thread threadA3 = new Thread();
    	// 启动线程
        threadA1.start();
        threadA2.start();
        threadA3.start();
    	// 卡死线程 等到A1 A2 A3都执行完毕后 再执行下方代码
        threadA1.join();
        threadA2.join();
        threadA3.join();
    	
    	// 声明B组线程 
    	Thread threadB1 = new Thread();
    	Thread threadB2 = new Thread();
    	// 启动B组线程
    	threadB1.start();
    	threadB2.start();
    }

--------------------

### 使用CountDownLatch ###

CountDownLatch 内核心方法:

*  await(): 等待. 当count值为0时, 才能继续执行.
 *  countDown(): 和方法名一样. 具体作用时将方法名称-1.
 *  getCount(): 获取当前count值.

### 简单都小测试 ###

import java.util.concurrent.CountDownLatch;
    
    public class TestJava {
    
        public static final CountDownLatch countDownLatch = new CountDownLatch(3);
    
        public static void main(String[] args) throws InterruptedException {
            Thread threadA1 = new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("AThreadName: "+Thread.currentThread().getName()+" - "+ "count:"+ countDownLatch.getCount());
                    countDownLatch.countDown();
                }
            });
            Thread threadA2 = new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("AThreadName: "+Thread.currentThread().getName()+" - "+ "count:"+ countDownLatch.getCount());
                    countDownLatch.countDown();
                }
            });
            Thread threadA3 = new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("AThreadName: "+Thread.currentThread().getName()+" - "+ "count:"+ countDownLatch.getCount());
                    countDownLatch.countDown();
                }
            });
            threadA1.start();
            threadA2.start();
            threadA3.start();
            // 等待3个线程都执行完成
            countDownLatch.await();
            Thread threadB1 = new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("BThreadName: "+Thread.currentThread().getName()+" - "+ "count:"+ countDownLatch.getCount());
                }
            });
            Thread threadB2 = new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println("BThreadName: "+Thread.currentThread().getName()+" - "+ "count:"+ countDownLatch.getCount());
                }
            });
            threadB1.start();
            threadB2.start();
        }
    
    
    }

*  测试结果  
    感兴趣的小伙伴可以本地运行下这段程序. 笔者运行这段程序结果如下所示:

AThreadName: Thread-0 - count:3
    AThreadName: Thread-1 - count:2
    AThreadName: Thread-2 - count:1
    BThreadName: Thread-3 - count:0
    BThreadName: Thread-4 - count:0

可以看到. A线程解锁的程序先执行. B线程解锁的程序后执行. 这样就能控制A任务组执行完成后, 才执行B任务组. 这在开发过程中也经常用到.

比如你的产品有2个步骤: 1. 验证数据 2. 插入数据. 当数据时成批处理的时候, 我们一般会希望任务1执行完成后, 再进行任务2的处理. 此时使用`CountDownLatch`准没错. 当然, 此工具类也有一定的局限性. 就是Count只能删除, 并不能重置. 这点, 我么在下一章`CycleBarrier`中再提及这点.

--------------------

### 实现细节 ###

CountDownLatch其实也是使用AQS对Java中锁的一种实现方式. 有空的话可以详细解读下具体实现.

package java.util.concurrent;
    import java.util.concurrent.locks.AbstractQueuedSynchronizer;

从包名可以看出. 其时属于`java.util.concurrent`包下的. 属于官方的提供的一个工具包.

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5Zyo6aOO5Lit55qE5oSP5b-X_size_20_color_FFFFFF_t_70_g_se_x_16]  
我们再来看下类图. 这个类非常简单, 无继承类. 基础的时默认基类`Object`. 其中的核心方法主要就是3个. `await()`/`await(long, TimeUnit)`后面这个应该是设置了超时时间, 为了防止死锁. `countDown()`减少count的值.

随后. 我们可以看下具体实现.

*  局部类Sync. 其主要实现了AQS.

/**
         * Synchronization control For CountDownLatch.
         * Uses AQS state to represent count.
         */
        private static final class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = 4982264981922014374L;
    		
    		// 通过count记录状态值 主要分为0和非0 这2种情况
            Sync(int count) {
                setState(count);
            }
    
            int getCount() {
                return getState();
            }
    		
    		// 主要重写方法1 当state 也就是count为0时才能获取Shared锁
            protected int tryAcquireShared(int acquires) {
                return (getState() == 0) ? 1 : -1;
            }
    		
    		// 主要重写方法2 
            protected boolean tryReleaseShared(int releases) {
                // Decrement count; signal when transition to zero
                for (;;) {
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    // 使用cas 让c=c-1 .
                    if (compareAndSetState(c, nextc))
                    	// 成功-1. 但是返回前再判断下当前锁是否已经停止了
                        return nextc == 0;
                }
            }
        }

*  其余方法

private final Sync sync;
        // 构造方法
        public CountDownLatch(int count) {
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
        }
        // 等待
            public void await() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
        // 等待 并设置超时时间
            public boolean await(long timeout, TimeUnit unit)
            throws InterruptedException {
            return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
        }
        // 让count值-1
            public void countDown() {
            sync.releaseShared(1);
        }
        // 获取当前count值. 在上方可以看到是通过getState()方法进行实现的.
            public long getCount() {
            return sync.getCount();
        }
        // 重写父类 toString方法 没啥意思
            public String toString() {
            return super.toString() + "[Count = " + sync.getCount() + "]";
        }

--------------------

### 后记![在这里插入图片描述][57e9ee74be6e4bd8b9690bd8986f5397.png] ###

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5Zyo6aOO5Lit55qE5oSP5b-X_size_9_color_FFFFFF_t_70_g_se_x_16]  
顺便随便看了一眼Java的这些工具包. 好家伙,还真不少. 有机会都撸一遍吧.

--------------------

### Reference ###

\[1\] JDK核心源码: java.util.concurrent.CountDownLatch

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5Zyo6aOO5Lit55qE5oSP5b-X_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/49bfaf855a95450ca270e98e2f4ffb4c.png
[57e9ee74be6e4bd8b9690bd8986f5397.png]: /images/20220828/1a388e033e9144acb8c564a699f5f719.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5Zyo6aOO5Lit55qE5oSP5b-X_size_9_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/bdb115ecf9e249529002bffefd5832ad.png