Semaphore信号量
简介
它本质上是一个共享锁,限制访问公共资源的线程数目,它也被称为计数信号量
acquire()许可一个线程, Semaphore – 1; 没有可用的许可时,Semaphore=0 ,线程阻塞
release()释放一个线程, Semaphore + 1
示例
public class MySemaphore {
public static void main(String[] args) {
// 使用线程池
ExecutorService exec = Executors.newCachedThreadPool();
// 只允许3个线程同时访问
final Semaphore semp = new Semaphore(3);
// 模拟4个客户端访问
for (int index = 0; index < 4; index++) {
Runnable run = new Runnable() {
public void run() {
try {
// 获取许可
semp.acquire();
System.out.println("线程"+ Thread.currentThread().getName() + "获得许可：");
// 模拟耗时的任务
for (int i = 0; i < 999999; i++);
// 释放许可
semp.release();
System.out.println("线程"+ Thread.currentThread().getName() + "释放许可：");
System.out.println("当前允许进入的任务个数："+ semp.availablePermits());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
exec.execute(run);
}
// 关闭线程池
exec.shutdown();
}
}控制台输出:
线程pool-1-thread-1获得许可：
线程pool-1-thread-2获得许可：
线程pool-1-thread-2释放许可：
当前允许进入的任务个数：2 //总共允许3个许可, 获取两个许可, 释放一个许可, 剩余2个许可
线程pool-1-thread-1释放许可：
当前允许进入的任务个数：2 //释放一个许可, 应该打印出1, 可以看出, Semaphore并不保证线程安全
线程pool-1-thread-3获得许可：
线程pool-1-thread-3释放许可：
当前允许进入的任务个数：2
线程pool-1-thread-4获得许可：
线程pool-1-thread-4释放许可：
当前允许进入的任务个数：3
CyclicBarrier 障碍器
简介
允许一组线程互相等待,到达一个公共的障碍点, 该组任务完成后, 再去完成另外一个任务
在释放等待线程后可以重用,它是循环的barrier
示例
public class MyCyclicBarrier {
public static void main(String[] args) {
//创建CyclicBarrier对象， 并设置执行完一组5个线程的并发任务后，再执行MainTask任务
CyclicBarrier cb = new CyclicBarrier(5, new MainTask());
new SubTask("A", cb).start();
new SubTask("B", cb).start();
new SubTask("C", cb).start();
new SubTask("D", cb).start();
new SubTask("E", cb).start();
}
}
/** 最后执行的任务 */
class MainTask implements Runnable {
public void run() {
System.out.println("......终于要执行最后的任务了......");
}
}
/** 一组并发任务 */
class SubTask extends Thread {
private String name;
private CyclicBarrier cb;
SubTask(String name, CyclicBarrier cb) {
this.name = name;
this.cb = cb;
}
public void run() {
System.out.println("[并发任务" + name + "] 开始执行");
for (int i = 0; i < 999999; i++); // 模拟耗时的任务
System.out.println("[并发任务" + name + "] 执行完毕，通知障碍器");
try {
// 每执行完一项任务就通知障碍器
cb.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
}控制台输出:
[并发任务A] 开始执行
[并发任务D] 开始执行
[并发任务C] 开始执行
[并发任务B] 开始执行
[并发任务E] 开始执行
[并发任务B] 执行完毕，通知障碍器
[并发任务E] 执行完毕，通知障碍器
[并发任务D] 执行完毕，通知障碍器
[并发任务A] 执行完毕，通知障碍器
[并发任务C] 执行完毕，通知障碍器
......终于要执行最后的任务了...... //可以看出执行一组任务后,在执行这个线程任务
CountDownLatch 障碍器
简介
允许1或N个线程等待其他线程完成后在执行
调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置
示例
public class MyCountDownLatch {
public static void main(String[] args) {
//启动会议室线程，等待与会人员参加会议
Conference conference = new Conference(3);
new Thread(conference).start();
//参会者线程
for(int i = 0 ; i < 3 ; i++){
Participater participater = new Participater("" + i , conference);
Thread thread = new Thread(participater);
thread.start();
}
}
}
/** 会场类 */
class Conference implements Runnable{
private final CountDownLatch countDown;//障碍器
public Conference(int count){
countDown = new CountDownLatch(count);
}
/** 与会人员到达 */
public void arrive(String name){
System.out.println(name + "到达.....");
//到达一个,锁计数器 - 1, 在计数到达0之前会一直阻塞
countDown.countDown();
System.out.println("还有 " + countDown.getCount() + "位没有到达...");
}
@Override
public void run() {
System.out.println("准备开会，参加会议人员总数为：" + countDown.getCount());
//调用await(),等待所有的与会人员到达
try {
countDown.await();
} catch (InterruptedException e) {
}
System.out.println("所有人员已经到达，会议开始.....");
}
}
/** 参会者类*/
class Participater implements Runnable{
private String name;
private Conference conference;
public Participater(String name,Conference conference){
this.name = name;
this.conference = conference;
}
@Override
public void run() {
conference.arrive(name);
}
}控制台输出:
准备开会，参加会议人员总数为：3
2到达.....
还有 2位没有到达...
0到达.....
还有 1位没有到达...
1到达.....
所有人员已经到达，会议开始.....
还有 0位没有到达...
Phaser
简介
推荐阅读: http://whitesock.iteye.com/blog/1135457
http://www.2cto.com/kf/201611/560952.html
任务数目是可变的: 可以在任何时间注册新的参与者;并且在抵达屏障点时,可以注销已经注册的参与者
phase和party
phase就是阶段，初值为0:
当所有的线程执行完本轮任务，同时开始下一轮任务时，意味着当前阶段已结束，
进入到下一阶段，phase的值自动加1
party就是线程: party=4就意味着Phaser对象当前管理着4个线程
boolean onAdvance(int phase, int registeredParties) :
1.当此方法返回true时，意味着Phaser被终止, 若此方法返回值为 phase>=3，其含义为当整个线程执行了4个阶段后，程序终止
2.当每一个阶段执行完毕，此方法会被自动调用 ,此方法内的代码会在每个阶段执行完毕时执行
示例: 可变数目的任务
import java.util.Random;
import java.util.concurrent.Phaser;
import java.util.concurrent.atomic.AtomicInteger; /**
*可变数目: 动态注册和取消
*
*示例:
* 在旅游过程中,有可能很凑巧遇到几个朋友,
* 然后他们听说你们在旅游,所以想要加入一起继续接下来的旅游.
* 也有可能,在旅游过程中,突然其中有某几个人临时有事,想退出这次旅游了
*/
public class MyPhaser_5 {
public static void main(String[] args) {
final int num = 3;
Phaser phaser = new Phaser(num){
/**
* 如果该方法返回true，那么Phaser会被终止, 默认实现是在注册任务数为0时返回true
* phase : 阶段数
* registeredParties : 注册的线程数
*/
@Override
protected boolean onAdvance(int phase, int registeredParties) {
System.out.println("" + getArrivedParties() + "个人都到齐了,第" + (phase + 1) + "次集合 \n");
return phase >= num;
}
}; new Thread(new TourismRunnable(phaser),"小明").start();
new Thread(new TourismRunnable(phaser),"小刚").start();
new Thread(new TourismRunnable(phaser),"小红").start();
}
} /** 旅行线程 */
class TourismRunnable implements Runnable{
Phaser phaser;
/**
* 每个线程保存一个朋友计数器,小红第一次遇到一个朋友,取名`小红的朋友0号`,第二次遇到一个朋友,取名为`小红的朋友1号`
*/
AtomicInteger frientCount = new AtomicInteger(); public TourismRunnable(Phaser phaser) {
this.phaser = phaser;
} @Override
public void run() {
switch (phaser.getPhase()){
case 0:if(!goToPoint("出发点")) break;
case 1:if(!goToPoint("旅游景点")) break;
case 2:if(!goToPoint("酒店")) break;
}
} /**
* @param point 目的地
* @return 返回true,说明还要继续旅游,否则就临时退出了
*/
private boolean goToPoint(String point){
try {
if(!randomEvent()){
//取消注册
phaser.arriveAndDeregister();
return false;
}
System.out.println(Thread.currentThread().getName() + "到了" + point); //阻塞
phaser.arriveAndAwaitAdvance();
return true;
} catch (Exception e) {
e.printStackTrace();
}
return false;
} /**
* 随机事件: 遇到新朋友一起旅游 或者 中途退出旅游
* @return 返回true,说明还要继续旅游,否则就临时退出了
*/
private boolean randomEvent() {
int random = new Random().nextInt(100);
String name = Thread.currentThread().getName(); if (random < 10){
int friendNum = 1;
System.out.println("=====================" + name + ":遇到了"+friendNum+"个朋友,要一起去旅游"); new Thread(new TourismRunnable(phaser), name + "的朋友" + frientCount.incrementAndGet() + "号").start();
//注册
phaser.bulkRegister(friendNum); }else if(random > 80){
System.out.println("=====================" + name + ":突然有事要离开一下,不和他们继续旅游了");
return false;
} return true;
}
}