欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

关于使用CyclicBarrier使主线程等待子线程执行完之后再向下执行的问题

程序员文章站 2022-06-10 23:33:25
...
CyclicBarrier
线程障碍,其实就是为线程制作一个集合点,相关知识网上一堆。

这次主要记录一下使用CyclicBarrier遇到的一点小问题。

需求:在主线程中创建两个子线程,希望能在子线程执行完成之后再执行主线程中的剩余代码。

代码一:

package com.iteye.wwwcomy.thread;

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

public class RaceCondition implements Runnable {
Counter counter;
final static CyclicBarrier cb = new CyclicBarrier(2, new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "执行");
}
});

RaceCondition() {
}

RaceCondition(Counter counter) {
this.counter = counter;
}

/**
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
Counter counter = new Counter();
ExecutorService service = Executors.newCachedThreadPool();
Thread t1 = new Thread(new RaceCondition(counter));
Thread t2 = new Thread(new RaceCondition(counter));
service.execute(t1);
service.execute(t2);
service.shutdown();
System.out.println(counter.count);
}

@Override
public void run() {
for (int i = 0; i < 5000000; i++)
counter.add(1);
try {
System.out.println("before");
cb.await();
System.out.println("complete");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
} finally {
}
}
}

class Counter {

/**
* 此处加上volatile并没有起到预期作用,依旧要把方法同步
* NOTICE 这里count的值的范围是多少?
* 循环十次的极限情况:
* 1.线程一拿到了0,等待线程二
* 2.线程二拿到了0,并且计算了9次,得到结果9并且写了回去
* 3.线程一第一次计算完,并且把1写入
* 4.线程二第十次取到了1
* 5.线程一计算完成,把结果写入
* 6.线程二把第十次的计算结果2写入
*/
protected volatile long count = 0;

public void add(long value) {
this.count++;
}
}


结果发现:[b]主线程依旧在子线程之前执行到了[/b]。

再看代码二:

package com.iteye.wwwcomy.thread;

import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RaceConditionWithCyclicBarrier implements Callable<Boolean> {
Counter2 counter;
static CyclicBarrier cb = null;

RaceConditionWithCyclicBarrier() {
}

RaceConditionWithCyclicBarrier(Counter2 counter) {
this.counter = counter;
}

/**
* @param args
* @throws Throwable
*/
public static void main(String[] args) throws Throwable {
Counter2 counter = new Counter2();

cb = new CyclicBarrier(2, new Runnable() { // 等到线程到达后执行一个后续task
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "a");
}
});

ArrayList<Callable<Boolean>> listCall = new ArrayList<Callable<Boolean>>();
listCall.add(new RaceConditionWithCyclicBarrier(counter));
listCall.add(new RaceConditionWithCyclicBarrier(counter));

ExecutorService executor = Executors.newFixedThreadPool(2); // 必须是allThread的个数
try {
executor.invokeAll(listCall);
} catch (Throwable e) {
executor.shutdown();
throw e;
} finally {
System.out.println("over");
}
executor.shutdown();
// 这句是在所有线程都跑完之后才会执行
System.out.println(counter.count);
}

@Override
public Boolean call() throws Exception {
for (int i = 0; i < 500000; i++)
counter.add(1);
cb.await();
return true;
}
}

class Counter2 {

/**
* 此处加上volatile并没有起到预期作用,依旧要把方法同步
* NOTICE 这里count的值的范围是多少?
* 循环十次的极限情况:
* 1.线程一拿到了0,等待线程二
* 2.线程二拿到了0,并且计算了9次,得到结果9并且写了回去
* 3.线程一第一次计算完,并且把1写入
* 4.线程二第十次取到了1
* 5.线程一计算完成,把结果写入
* 6.线程二把第十次的计算结果2写入
*/
protected volatile long count = 0;

public void add(long value) {
this.count++;
}
}


竟然发现这次主线程在executor.invokeAll(listCall);后被挂起了,直到子线程执行完成之后才执行。

其实就是使用Runnable和Callable之后executor的API使用不同而已,怎么结果差距这么大。

断个点,在visualVM里面dump了一下线程发现Callable的executor执行代码中将主线程挂起了,见线程堆栈,具体源码按照堆栈进去跟就是了:

[quote]"pool-1-thread-2" prio=6 tid=0x00000000067f7800 nid=0x1534 at breakpoint[0x000000000876f000]
java.lang.Thread.State: RUNNABLE
at com.iteye.wwwcomy.thread.RaceConditionWithCyclicBarrier.call(RaceConditionWithCyclicBarrier.java:60)
at com.iteye.wwwcomy.thread.RaceConditionWithCyclicBarrier.call(RaceConditionWithCyclicBarrier.java:1)
at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
at java.util.concurrent.FutureTask.run(FutureTask.java:138)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

Locked ownable synchronizers:
- <0x00000007d5f2ca10> (a java.util.concurrent.locks.ReentrantLock$NonfairSync)

"pool-1-thread-1" prio=6 tid=0x00000000067f6800 nid=0x1da4 at breakpoint[0x000000000866f000]
java.lang.Thread.State: RUNNABLE
at com.iteye.wwwcomy.thread.RaceConditionWithCyclicBarrier.call(RaceConditionWithCyclicBarrier.java:60)
at com.iteye.wwwcomy.thread.RaceConditionWithCyclicBarrier.call(RaceConditionWithCyclicBarrier.java:1)
at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
at java.util.concurrent.FutureTask.run(FutureTask.java:138)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

Locked ownable synchronizers:
- <0x00000007d5f2c7d8> (a java.util.concurrent.locks.ReentrantLock$NonfairSync)

"main" prio=6 tid=0x00000000002ae800 nid=0x168c waiting on condition [0x000000000251f000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x00000007d5f2c770> (a java.util.concurrent.FutureTask$Sync)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:156)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:811)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:969)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1281)
at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:218)
at java.util.concurrent.FutureTask.get(FutureTask.java:83)
at java.util.concurrent.AbstractExecutorService.invokeAll(AbstractExecutorService.java:205)
at com.iteye.wwwcomy.thread.RaceConditionWithCyclicBarrier.main(RaceConditionWithCyclicBarrier.java:46)

Locked ownable synchronizers:
- None[/quote]

可见只有代码二的做法才能使主线程在子线程完成之后再执行,CyclicBarrier自身并不保证主线程在子线程完成之后执行。


[b][color=red]CyclicBarrier仅仅是为调用await方法的线程设置一个集合点![/[/color]b]

[b]代码一的写法其实稍微做一点改变,也可以实现效果:
1.修改CyclicBarrier的初始线程数+1,
2.在主线程里面也加入 cb.await(); 就可以了[/b]

当时先写的代码二,所以有点先入为主了。。本来正在研究volatile,遇到这个问题先记录一下。


另外,CyclicBarrier的第二个参数,是使用最后一个执行完的线程执行的。

----2017.05.10补充-------------
最近在面试,又回看了一下当时写的这个blog,实际上代码二中主线程在子线程之后执行的原因是调用的executor方法不同,这个在blog里面没提出来,
代码一里面使用的是(submit方法同样,区别见http://*.com/questions/3929342/choose-between-executorservices-submit-and-executorservices-execute)
executor.execute()

代码二里面使用的是
executor.invokeAll()

代码二中会在子线程执行完之后才执行主线程的后续内容。代码一会直接执行主线程后续内容。