前言
在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包下的. 属于官方的提供的一个工具包.
我们再来看下类图. 这个类非常简单, 无继承类. 基础的时默认基类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() + "]";
}
后记
顺便随便看了一眼Java的这些工具包. 好家伙,还真不少. 有机会都撸一遍吧.
Reference
[1] JDK核心源码: java.util.concurrent.CountDownLatch