并发工具类
Semaphore限制线程并发数
- 其构造参数new Semaphore(10,true)是初始允许的线程数量，true/false决定是否公平信号量
- accquire(2) 动态添加许可数量
- acquireUninterruptibly() 使等待进入acquire()方法的线程不允许被中断
- availablePermits() 返回Semaphore对象中当前可用许可数
- drainPermits() 可获取并返回立即可用的所有许可个数，并将可用许可置0
- getQueueLength() 取得等待许可的线程个数
- hasQueuedThreads()判断有没有线程再等待这个许可
- tryAcquire() 尝试获得一个许可，如果获取不到返回false.添加参数，尝试获得X个许可
- tryAcquire(long timeout,TimeUnit unit) 在指定时间内尝试获取一个许可，如果获取不到返回false
- tryAcquire(int permits,long timeout,TimeUnit unit)在指定时间内尝试获取X个许可，如果获取不到返回false
public class Service{
//同一时间最初最多允许2个线程acquire和release之间的代码
private Semaphore semaphore = new Semaphore(2);
public void testMethod(){
try{
semaphore.acquire();//acquire(int permits)每调用一次就使用多少个许可
System.out.println(Thread.currentThread().getName()+"begin timer"+System.currentTimeMillis());
semaphore.release();
System.out.println(Thread.currentThread().getName()+"end timer"+System.currentTimeMillis());
}catch(InterruptedException e){
e.printStackTrace();
}
}
public static void main(String[] args){
Service service = new Service();
ThreadA a = new ThreadA(service);
a.setName("A");
ThreadB b = new ThreadB(service);
b.setName("B");
ThreadC c = new ThreadC(service);
c.setName("C");
a.start();
b.start();
c.start();
}
}
//创建三个线程(略)
public class ThreadA extends Thread{
private Service service;
public ThreadA(Service service){
super();
this.service = service;
}
@Override
public void run(){
service.testMethod();
}
}
Exchanger两个线程之间交换数据
- exchange() 等待其他线程来取得数据，如果没有其他线程取得数据，则一直阻塞等待。
- exchange(V x,long timeout,TimeUnit unit)在指定的时间内没有其他线程获取数据，超时异常
public class ThreadA extends Thread{
private Exchange<String> exchange;
public ThreadA(Exchanger<String> exchanger){
super();
this.exchanger = exchanger;
}
public void run(){
try{
System.out.println("在线程A中得到线程B的值="+exchanger.exchange("中国人A"));
System.out.println("A end!");
}catch(){}
}
}
public class ThreadB extends Thread{
private Exchange<String> exchange;
public ThreadA(Exchanger<String> exchanger){
super();
this.exchanger = exchanger;
}
public void run(){
try{
System.out.println("在线程B中得到线程A的值="+exchanger.exchange("中国人B"));
System.out.println("A end!");
}catch(){}
}
}
public class Run{
public static void main(String[] args){
Exchanger<String> exchanger = new Exchanger<>();
ThreadA a = new ThreadA(exchanger);
ThreadB b = new ThreadB(exchanger);
a.start();
System.out.println("main end!");
}
}
CountDownLatch 线程以组团的方式执行任务
- await()判断计数是否为0，如果不为0则等待，
- countDown()方法，其他线程调用，计数减1，当计数减到0时，等待的线程继续运行
- getCount() 获取当前的计数个数
计数无法被重置，需要重置计数，使用CyclicBarrier
public class MyService{
private CountDownLatch down = new CountDownLatch(1);
public void testMethod(){
try{
System.out.println("A");
down.await();//判断计数是否为0，不为0就等待
System.out.println("B");
}catch(InterruptedException e){
e.printStackTrace();
}
}
public void downMethod(){
System.out.println("x");
down.countDown();//其他线程调用将计数减1
}
}
public class MyThread extends Thread{
private MyService myService;
public MyThread(MyService myService){
super();
this.myService = myService;
}
public void run(){
myService.testMethod();
}
}
public class Run{
public static void main(String[] args){
MyService service = new MyService();
MyThread t = new MyThread(service);
t.start();
Thread.sleep(2000);
service.downMethod();
}
}
public class MyThread extends Thread{
private CountDownLatch maxRunner;
public MyThread(CountDownLatch maxRunner){
super();
this.maxRunner = maxRunner;
}
public void run(){
try{
Thread.sleep(2000);
maxRunner.countDown();//计数减1
}catch(Exception e){
e.printStackTrace();
}
}
}
public class Run{
public static void main(){
//初始化为10
CountDownLatch maxRunner = new CountDownLatch(10);
MyThread[] tArray = new MyThread[Integer.parseInt(""+maxRunner.getCount())];
//启动十个线程
for(int i=0;i<tArray.length;i++){
tArray[i] = new MyThread(maxRunner);
tArray[i].setName("线程"+(i+1));
tArray[i].start
}
//直到子线程全部执行，主线程才结束阻塞
maxRunner.await();
System.out.println("线程全部执行");
}
}
CyclicBarrier循环地实现一起做任务的的目标
一组线程互相等待直到达到某个公共屏障点，其公共屏障点可以重用。
- getNumberWaiting() 获取有几个线程已经到达屏障点
- isBroken() 查询此屏障是否处于损坏状态
- await(long timeout,TimeUnit unit) 指定的时间内达到Parties数量则继续向下运行，否则出现超时，则抛出TimeoutException异常。cyclicBarrier.await(2,TimeUnit.SECONDS);
- getNumverWaiting() 有几个线程已经到达了屏障点
- getParties() 取得parties个数
- reset() 重置屏障，线程会出现Broken异常
public class MyThread extends Thread{
private CyclicBarrier cb;
public MyThread(CyclicBarrier cb){
super();
this.cb = cb;
}
public void run(){
try{
Thread.sleep((int)(Math.random()*1000));
System.out.println(Thread.currentThread().getName()+"到了"+System.currentTimeMillis());
cb.await();//等待（5个线程都执行了该await方法才能继续执行）
}catch(Exception e){
e.printStackTrace();
}
}
}
public class Run{
public static void main(String[] args){
CyclicBarrier cb = new CyclicBarrier(5,new Runnable(){
public void run(){
System.out.println("全部到齐");
}
});
//创建5个子线程
MyThread[] threadArray = new MyThread[5];
for(int i=0;i<threadArray.length;i++){
threadArray[i] = new MyThread(cb);
}
//启动
for(int i=0;i<threadArray.length;i++){
threadArray[i].start();
}
}
}
public class MyThread extends Thread{
private CyclicBarrier cb;
public MyThread(CyclicBarrier cb){
super();
this.cb = cb;
}
public void run(){
try{
System.out.println(Thread.currentThread().getName()+"等待凑齐2个继续运行"+System.currentTimeMillis());
cb.await();
System.out.println(Thread.currentThread().getName()+"已经凑齐两个"+System.currentTimeMillis());
}catch(Exception e){
e.printStackTrace();
}
}
}
public class Test{
public static void main(String[] args){
CyclicBarrier cb = new CyclicBarrier(2,new Runnable(){
public void run(){
System.out.println("全来了");
}
});
for(int i=0;i<4;i++){
ThreadA threadA = new ThreadA(cb);
threadA.start();
Thread.sleep(2000);
}
}
}
public class ThreadA extends Thread{
private CyclicBarrier cb;
public ThreadA(CyclicBarrier cb){
this.cb = cb;
}
public void run(){
try{
cb.await();
}catch(Exception e){
e.printStackTrace();
}
}
}
public class Test{
public static void main(String[] args){
CyclicBarrier cb = new CyclicBarrier(3);
ThreadA threadA1 = new ThreadA(cb);
threadA1.start();
Thread.sleep(500);
System.out.println(cb.getNumberWaiting());//1
ThreadA threadA2 = new ThreadA(cb);
threadA2.start();
Thread.sleep(500);
System.out.println(cb.getNumberWaiting());//2
ThreadA threadA3 = new ThreadA(cb);
threadA3.start();
Thread.sleep(500);
System.out.println(cb.getNumberWaiting());//0 到达屏障点的线程为0，即重置
ThreadA threadA4 = new ThreadA(cb);
threadA4.start();
Thread.sleep(500);
System.out.println(cb.getNumberWaiting());//1
}
}
CyClicBarrier与CountDownLatch的区别：
- CountDownLatch：一个线程或者多个线程，等待另外一个线程或多个线程完成某个事情之后才能继续执行
- CyClicBarrier：多个线程之间相互等待，任何一个线程完成之前，所有的线程都必须等待。重点是其中任何一个线程没有完成，则所有线程都必须等待。（计数属于加法操作）
Phaser JDK7新增 对计数器的操作是加法操作 设置多屏障的功能
- arriveAndAwaitAdvance() 计数不足，线程呈阻塞状态，不能继续向下运行
- arriveAndDeregister() 当前线程退出，使parties减1
- getPhase() 获取已经到达第几个屏障了
- onAdvance() 通过新的屏障时被调用
- getRegisteredParties() 获得注册的parties数量
- register() 每执行一次就动态添加一个parties值
- bulkRegister() 批量增加parties数量（参数中添加需要增加的数量）
- getArrivedParties() 获得未被使用的parties个数
- getUnarrivedParties()
- arrive() 使parties值加1，并且不在屏障出等待，直接向下面的代码继续运行
- awaitAdvance(int phase) 如果传入参数phase值和当前getPhase()方法返回值一样，则在屏障处等待否则向下运行。
- awaitAdvanceInterruptibly(int) 不可中断
- awaitAdvanceInterruptibly(int,long,TimeUnit)指定的栏数等待最大的单位时间，如果在指定的时间内栏数未变，则出现异常，否则向下运行
- forceTermination() 使Phaser对象的屏障功能失效
- isTerminated() 判断Phaser对象是否呈销毁状态
public class PrintTools{
public static Phaser phaser;
public static void methodA(){
//A1 begin
phaser.arriveAndAwaitAdvance();
//A1 end
//A2 begin
phaser.arriveAndAwaitAdvance();
//A2 end
}
public static void methodA(){
//A1 begin
Thread.sleep(5000);
phaser.arriveAndAwaitAdvance();
//A1 end
//A2 begin
Thread.sleep(5000);
phaser.arriveAndAwaitAdvance();
//A2 end
}
}
//定义三个线程A,B,C 三个线程的run方法调用上面的methodA和methodB （略）
//主类中Phaser phaser = new Phaser(3)，并启动三个线程
public static void method1(){
//...业务处理
System.out.println(phaser.getRegisteredParties());
phaser.arriveAndDeregister();//使parties减1
System.out.println(phaser.getRegisteredParties());
}
public void run(){
phaser.arriveAndAwaitAdvance();
phaser.getPhase();//1
phaser.arriveAndAwaitAdvance();
phaser.getPhase();//2
}
//onAdvance() 通过新的屏障被调用
Phaser phaser = new Phaser(2){
protected boolean onAdvance(int phase,int registeredParties){
return true;//返回true，不等待了，Phaser呈无效、销毁状态
//false 则Phaser继续工作
}
};
投递美团等公司10个岗位