并发工具类同步屏障CyclicBarrier

  categories:资料  author:

简介

CyclicBarrier 的字面意思是可循环使用(Cyclic)的屏障(Barrier)。它要做的事情是,让一组线程到达一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续干活。CyclicBarrier默认的构造方法是CyclicBarrier(int parties),其参数表示屏障拦截的线程数量,每个线程调用await方法告诉CyclicBarrier我已经到达了屏障,然后当前线程被阻塞。

package cn.itcast.heima2;
实例代码如下:
public class CyclicBarrierTest {static CyclicBarrier c = new CyclicBarrier(2);public static void main(String[] args) {
new Thread(new Runnable() {

@Override
public void run() {
try {
c.await();
} catch (Exception e) {

}
System.out.println(1);
}
}).start();

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

}
System.out.println(2);
}
}

输出
2
1

或者输出
1
2

如果把new CyclicBarrier(2)修改成new CyclicBarrier(3)则主线程和子线程会永远等待,因为没有第三个线程执行await方法,即没有第三个线程到达屏障,所以之前到达屏障的两个线程都不会继续执行。

CyclicBarrier还提供一个更高级的构造函数CyclicBarrier(int parties, Runnable barrierAction),用于在线程到达屏障时,优先执行barrierAction,方便处理更复杂的业务场景。代码如下:

public class CyclicBarrierTest2 {

static CyclicBarrier c = new CyclicBarrier(2, new A());

public static void main(String[] args) {
new Thread(new Runnable() {

@Override
public void run() {
try {
c.await();
} catch (Exception e) {

}
System.out.println(1);
}
}).start();

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

}
System.out.println(2);
}

static class A implements Runnable {

@Override
public void run() {
System.out.println(3);
}

}

}

输出
3
1
2

CyclicBarrier的应用场景

CyclicBarrier可以用于多线程计算数据,最后合并计算结果的应用场景。比如我们用一个Excel保存了用户所有银行流水,每个Sheet保存一个帐户近一年的每笔银行流水,现在需要统计用户的日均银行流水,先用多线程处理每个sheet里的银行流水,都执行完之后,得到每个sheet的日均银行流水,最后,再用barrierAction用这些线程的计算结果,计算出整个Excel的日均银行流水。

CyclicBarrier和CountDownLatch的区别

  • CountDownLatch的计数器只能使用一次。而CyclicBarrier的计数器可以使用reset() 方法重置。所以CyclicBarrier能处理更为复杂的业务场景,比如如果计算发生错误,可以重置计数器,并让线程们重新执行一次。
  • CyclicBarrier还提供其他有用的方法,比如getNumberWaiting方法可以获得CyclicBarrier阻塞的线程数量。isBroken方法用来知道阻塞的线程是否被中断。比如以下代码执行完之后会返回true。

isBroken的使用代码如下:

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest3 {

static CyclicBarrier c = new CyclicBarrier(2);

public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
Thread thread = new Thread(new Runnable() {

@Override
public void run() {
try {
c.await();
} catch (Exception e) {
}
}
});
thread.start();
thread.interrupt();
try {
c.await();
} catch (Exception e) {
System.out.println(c.isBroken());
}
}
}

输出

true

原创文章,转载请注明: 转载自并发编程网 – ifeve.com本文链接地址: 并发工具类(二)同步屏障CyclicBarrier

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

JUC回顾之-CyclicBarrier底层实现和原理

1.CyclicBarrier 字面意思是可循环(Cyclic)使用的屏障(Barrier)。它要做的事情是让一组线程到达一个屏障(同步点)时被阻塞,直到最后一个线程到达屏障时候,屏障才会开门。所有被屏障拦截的线程才会运行。

2.常用的方法:

CyclicBarrier(int parties)
创建一个新的 CyclicBarrier,它将在给定数量的参与者(线程)处于等待状态时启动,但它不会在启动 barrier 时执行预定义的操作。

CyclicBarrier(int parties, Runnable barrierAction)
创建一个新的 CyclicBarrier,它将在给定数量的参与者(线程)处于等待状态时启动,并在启动 barrier 时执行给定的屏障操作,该操作由最后一个进入 barrier 的线程执行。

int await()
在所有参与者都已经在此 barrier 上调用 await 方法之前,将一直等待。

int await(long timeout, TimeUnit unit)
在所有参与者都已经在此屏障上调用 await 方法之前将一直等待,或者超出了指定的等待时间。

int getNumberWaiting()
返回当前在屏障处等待的参与者数目。

int getParties()
返回要求启动此 barrier 的参与者数目。

boolean isBroken()
查询此屏障是否处于损坏状态。

void reset()
将屏障重置为其初始状态。如果调用了该函数,则在等待的线程将会抛出BrokenBarrierException异常。

3.底层原理实现

CyclicBarrier是由ReentrantLock可重入锁和Condition共同实现的。

具体实现源码如下:

//1.CyclicBarrier构造方法
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
// parties表示“必须同时到达barrier的线程个数”。
this.parties = parties;
// count表示“处在等待状态的线程个数”。
this.count = parties;
// barrierCommand表示“parties个线程到达barrier时,会执行的动作”。
this.barrierCommand = barrierAction;
}

1. await源码分析:

public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen;
}
}
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
final ReentrantLock lock = this.lock;
// 获取“独占锁(lock)”
lock.lock();
try {
// 保存“当前的generation”
final Generation g = generation;

// 若“当前generation已损坏”,则抛出异常。
if (g.broken)
throw new BrokenBarrierException();

// 如果当前线程被中断,则通过breakBarrier()终止CyclicBarrier,唤醒CyclicBarrier中所有等待线程。
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}

// 将“count计数器”-1
int index = –count;
// 如果index=0,则意味着“有parties个线程到达barrier”。
if (index == 0) {  // tripped
boolean ranAction = false;
try {
// 如果barrierCommand不为null,则执行该动作。
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
// 唤醒所有等待线程,并更新generation。
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}

// 当前线程一直阻塞,直到“有parties个线程到达barrier” 或 “当前线程被中断” 或 “超时”这3者之一发生,
// 当前线程才继续执行。
for (;;) {
try {
// 如果不是“超时等待”,则调用await()进行等待;否则,调用awaitNanos()进行等待。
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
// 如果等待过程中,线程被中断,则执行下面的函数。
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
Thread.currentThread().interrupt();
}
}

// 如果“当前generation已经损坏”,则抛出异常。
if (g.broken)
throw new BrokenBarrierException();

// 如果“generation已经换代”,则返回index。
if (g != generation)
return index;

// 如果是“超时等待”,并且时间已到,则通过breakBarrier()终止CyclicBarrier,唤醒CyclicBarrier中所有等待线程,并抛出TimeoutException异常
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
// 释放“独占锁(lock)”
lock.unlock();
}
}

实例如下:

1.await()方法使用

public class CyclicBarrierTest {

private static int SIZE = 5;

private static CyclicBarrier cb;

public static void main(String[] args) {
cb = new CyclicBarrier(SIZE);
for (int i = 0; i < SIZE; i++) {
new MyTask().start();
}

}

static class MyTask extends Thread {
@Override
public void run() {
try {

System.out.println(“线程” + Thread.currentThread().getName() + “正在执行同一个任务”);
// 以睡眠来模拟几个线程执行一个任务的时间
Thread.sleep(1000);
System.out.println(“线程” + Thread.currentThread().getName() + “执行任务完成,等待其他线程执行完毕”);
// 用来挂起当前线程,直至所有线程都到达barrier状态再同时执行后续任务;
cb.await();

} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println(“所有线程写入完毕”);

}

}

}

结果

线程Thread-0正在执行同一个任务
线程Thread-2正在执行同一个任务
线程Thread-1正在执行同一个任务
线程Thread-3正在执行同一个任务
线程Thread-4正在执行同一个任务
线程Thread-3执行任务完成,等待其他线程执行完毕
线程Thread-4执行任务完成,等待其他线程执行完毕
线程Thread-0执行任务完成,等待其他线程执行完毕
线程Thread-1执行任务完成,等待其他线程执行完毕
线程Thread-2执行任务完成,等待其他线程执行完毕
所有线程写入完毕
所有线程写入完毕
所有线程写入完毕
所有线程写入完毕
所有线程写入完毕

2.使用 barrier.await(2000, TimeUnit.MILLISECONDS)方法:

public class CyclicBarrierWriteDataWaitTimeOutTest {

private static final int THREAD_NUM = 5;

private static final Random random = new Random();

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

// 使用构造方法:public CyclicBarrier(int parties, Runnable barrierAction)
// 参数parties表示一共有多少线程参与这次“活动”,barrierAction是可选的,用来指定当所有线程都完成这些必须的“任务”之后要干的其他事情
CyclicBarrier barrier = new CyclicBarrier(THREAD_NUM, new Runnable() {

@Override
public void run() {
// 最后写完数据的线程,会执行这个任务
System.out.println(Thread.currentThread().getId() + “:所有线程写数据完毕!”);
}
});

// 启动5个线程,写数据
for (int i = 0; i < THREAD_NUM; i++) {
if (i < THREAD_NUM – 1) {
Thread t = new Thread(new MyTask(barrier));
t.start();
} else {
// 最后一个线程延迟3秒执行
Thread.sleep(3000);
Thread t = new Thread(new MyTask(barrier));
t.start();
}

}

}

/**
*
* (线程类)
*
* <p>
* 修改历史:                                            <br>
* 修改日期            修改人员       版本             修改内容<br>
* ————————————————-<br>
* 2016年8月26日 上午11:21:39   user     1.0        初始化创建<br>
* </p>
*
* @author        Peng.Li
* @version        1.0
* @since        JDK1.7
*/
static class MyTask extends Thread {

private CyclicBarrier barrier;

public MyTask(CyclicBarrier barrier) {
this.barrier = barrier;
}

@Override
public void run() {

int time = random.nextInt(1000);
System.out.println(Thread.currentThread().getId() + “:需要” + time + “毫秒的时间写入数据”);

try {
Thread.sleep(time);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
System.out.println(Thread.currentThread().getId() + “:写入数据完毕,等待其他线程写入数据”);
try {
// 用来挂起当前线程,直至所有线程都到达barrier状态再同时执行后续任务;
// 等待所有线程都调用过此函数才能继续后续动作
// 只等待2s,那么最后一个线程3秒才执行,则必定会超时
barrier.await(2000, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getId() + “:所有线程写入数据完毕,继续处理其他任务…”);

}

}

}

从结果分析:

可以看到,前面四个线程等待最后一个线程超时了,这个时候这四个线程不会再等待最后一个线程写入完毕,而是直接抛出BrokenBarrierException

异常,继续执行后续的动作。最后一个线程完成写入数据操作后也继续了后续的动作。

需要理解的是:最后一个线程发生超时的异常,其他的线程不会继续等待,而是去执行其他的任务。

Thread-1:需要650毫秒的时间写入数据
Thread-3:需要297毫秒的时间写入数据
Thread-5:需要755毫秒的时间写入数据
Thread-7:需要79毫秒的时间写入数据
Thread-7:写入数据完毕,等待其他线程写入数据
Thread-3:写入数据完毕,等待其他线程写入数据
Thread-1:写入数据完毕,等待其他线程写入数据
Thread-5:写入数据完毕,等待其他线程写入数据
Thread-7:所有线程写入数据完毕,继续处理其他任务…
Thread-1:所有线程写入数据完毕,继续处理其他任务…
java.util.concurrent.TimeoutException
Thread-3:所有线程写入数据完毕,继续处理其他任务…
Thread-5:所有线程写入数据完毕,继续处理其他任务…
at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:250)
at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:427)
at concurrentMy.CountAndCyclicAndSemaphore.cyclibarrier.test.CyclicBarrierWriteDataWaitTimeOutTest$MyTask.run(CyclicBarrierWriteDataWaitTimeOutTest.java:299)
at java.lang.Thread.run(Thread.java:745)
java.util.concurrent.BrokenBarrierException
at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:243)
at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:427)
at concurrentMy.CountAndCyclicAndSemaphore.cyclibarrier.test.CyclicBarrierWriteDataWaitTimeOutTest$MyTask.run(CyclicBarrierWriteDataWaitTimeOutTest.java:299)
at java.lang.Thread.run(Thread.java:745)
java.util.concurrent.BrokenBarrierException
at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:243)
at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:427)
at concurrentMy.CountAndCyclicAndSemaphore.cyclibarrier.test.CyclicBarrierWriteDataWaitTimeOutTest$MyTask.run(CyclicBarrierWriteDataWaitTimeOutTest.java:299)
at java.lang.Thread.run(Thread.java:745)
java.util.concurrent.BrokenBarrierException
at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:243)
at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:427)
at concurrentMy.CountAndCyclicAndSemaphore.cyclibarrier.test.CyclicBarrierWriteDataWaitTimeOutTest$MyTask.run(CyclicBarrierWriteDataWaitTimeOutTest.java:299)
at java.lang.Thread.run(Thread.java:745)
Thread-9:需要164毫秒的时间写入数据
Thread-9:写入数据完毕,等待其他线程写入数据java.util.concurrent.BrokenBarrierException
Thread-9:所有线程写入数据完毕,继续处理其他任务…

at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:200)
at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:427)
at concurrentMy.CountAndCyclicAndSemaphore.cyclibarrier.test.CyclicBarrierWriteDataWaitTimeOutTest$MyTask.run(CyclicBarrierWriteDataWaitTimeOutTest.java:299)
at java.lang.Thread.run(Thread.java:745)

Face your past without regret. Handle your present with confidence.Prepare for future without fear. keep the faith and drop the fear. 面对过去无怨无悔,把握现在充满信心,备战未来无所畏惧。保持信念,克服恐惧!一点一滴的积累,一点一滴的沉淀,学技术需要不断的积淀!

来源:https://www.cnblogs.com/200911/p/6060195.html



快乐成长 每天进步一点点      京ICP备18032580号-1