码迷,mamicode.com
首页 > 编程语言 > 详细

多线程并发常用类:condition,semaphore,CyclicBarrier,countdownlatch,exchanger使用整理

时间:2015-01-27 09:25:18      阅读:149      评论:0      收藏:0      [点我收藏+]

标签:

condition 类:

作为一个示例,假定有一个绑定的缓冲区,它支持 put 和 take 方法。如果试图在空的缓冲区上执行 take 操作,则在某一个项变得可用之前,线程将一直阻塞;如果试图在满的缓冲区上执行 put 操作,则在有空间变得可用之前,线程将一直阻塞。我们喜欢在单独的等待 set 中保存 put 线程和 take 线程,这样就可以在缓冲区中的项或空间变得可用时利用最佳规划,一次只通知一个线程。可以使用两个 Condition 实例来做到这一点。 

class BoundedBuffer {
   final Lock lock = new ReentrantLock();
   final Condition notFull  = lock.newCondition(); 
   final Condition notEmpty = lock.newCondition(); 

   final Object[] items = new Object[100];
   int putptr, takeptr, count;

   public void put(Object x) throws InterruptedException {
     lock.lock();
     try {
       while (count == items.length) 
         notFull.await();
       items[putptr] = x; 
       if (++putptr == items.length) putptr = 0;
       ++count;
       notEmpty.signal();
     } finally {
       lock.unlock();
     }
   }

   public Object take() throws InterruptedException {
     lock.lock();
     try {
       while (count == 0) 
         notEmpty.await();
       Object x = items[takeptr]; 
       if (++takeptr == items.length) takeptr = 0;
       --count;
       notFull.signal();
       return x;
     } finally {
       lock.unlock();
     }
   } 
 }


semaphore类:
信号量(Semaphore),有时被称为信号灯,是在多线程环境下使用的一种设施, 它负责协调各个线程, 以保证它们能够正确、合理的使用公共资源

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

/**
 * 信号量 
 * 
 * @author 林计钦
 * @version 1.0 2013-7-25 下午02:03:40
 */
public class SemaphoreTest {
    public static void main(String[] args) {
        // 线程池
        ExecutorService exec = Executors.newCachedThreadPool();
        // 只能5个线程同时访问
        final Semaphore semp = new Semaphore(5);
        // 模拟20个客户端访问
        for (int index = 0; index < 50; index++) {
            final int NO = index;
            Runnable run = new Runnable() {
                public void run() {
                    try {
                        // 获取许可
                        semp.acquire();
                        System.out.println("Accessing: " + NO);
                        Thread.sleep((long) (Math.random() * 10000));
                        // 访问完后,释放
                        semp.release();
                        //availablePermits()指的是当前信号灯库中有多少个可以被使用
                        System.out.println("-----------------" + semp.availablePermits()); 
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            exec.execute(run);
        }
        // 退出线程池
        exec.shutdown();
    }
}


CyclicBarrier介绍 (二)

张孝祥视频学习笔记:

CyclicBarrier 表示大家彼此等待,大家集合好后才开始出发,分散活动后又在i指定地点集合碰面,这就好比整个公司的人员利用周末时间集体郊游一样,先各自从家出发到公司集合后,再同时出发到公园游玩,在指定地点集合后再同时开始就餐……

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

public class CyclicBarrierTest {
 public static void main(String [] args){
  ExecutorService service=Executors.newCachedThreadPool();
  final CyclicBarrier cb=new CyclicBarrier(3);  //三个线程同时到达
  for(int i=0;i<3;i++){          
   Runnable runnable=new Runnable(){
    public void run(){
     try {
      Thread.sleep((long)(Math.random()*10000));
      System.out.println("线程"+Thread.currentThread().getName()+
        "即将到达集合地点1,当前已有"+(cb.getNumberWaiting()+1)+"个已到达"+
        (cb.getNumberWaiting()==2?"都到齐了,继续走啊":"正在等候"));
      try {
       cb.await();
      } catch (BrokenBarrierException e) {
       // TODO Auto-generated catch block
       e.printStackTrace();
      }
      Thread.sleep((long)(Math.random()*10000));
      System.out.println("线程"+Thread.currentThread().getName()+
        "即将到达集合地点2,当前已有"+(cb.getNumberWaiting()+1)+"个已到达"+
        (cb.getNumberWaiting()==2?"都到齐了,继续走啊":"正在等候"));
      try {
       cb.await();
      } catch (BrokenBarrierException e) {
       // TODO Auto-generated catch block
       e.printStackTrace();
      }
      Thread.sleep((long)(Math.random()*10000));
      System.out.println("线程"+Thread.currentThread().getName()+
        "即将到达集合地点3,当前已有"+(cb.getNumberWaiting()+1)+"个已到达"+
        (cb.getNumberWaiting()==2?"都到齐了,继续走啊":"正在等候"));
      try {
       cb.await();
      } catch (BrokenBarrierException e) {
       // TODO Auto-generated catch block
       e.printStackTrace();
      }
     } catch (InterruptedException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
     }
    }
   };
   service.execute(runnable);
  }
  service.shutdown();
 }
}

运行结果:

线程pool-1-thread-3即将到达集合地点1,当前已有1个已到达正在等候
线程pool-1-thread-2即将到达集合地点1,当前已有2个已到达正在等候
线程pool-1-thread-1即将到达集合地点1,当前已有3个已到达都到齐了,继续走啊
线程pool-1-thread-1即将到达集合地点2,当前已有1个已到达正在等候
线程pool-1-thread-2即将到达集合地点2,当前已有2个已到达正在等候
线程pool-1-thread-3即将到达集合地点2,当前已有3个已到达都到齐了,继续走啊
线程pool-1-thread-2即将到达集合地点3,当前已有1个已到达正在等候
线程pool-1-thread-1即将到达集合地点3,当前已有2个已到达正在等候
线程pool-1-thread-3即将到达集合地点3,当前已有3个已到达都到齐了,继续走啊



countdownlatch类:
/**
CountDownLatch类是一个同步计数器,构造时传入int参数,该参数就是计数器的初始值,每调用一次countDown()方法,计数器减1,计数器大于0 时,await()方法会阻塞程序继续执行
CountDownLatch如其所写,是一个倒计数的锁存器,当计数减至0时触发特定的事件。利用这种特性,可以让主线程等待子线程的结束。下面以一个模拟运动员比赛的例子加以说明。
*/
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CountDownLatchDemo {
private static final int PLAYER_AMOUNT = 5;
public CountDownLatchDemo() {
// TODO Auto-generated constructor stub
}
/**
*
@param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
//对于每位运动员,CountDownLatch减1后即结束比赛
CountDownLatch begin = new CountDownLatch(1);
//对于整个比赛,所有运动员结束后才算结束
CountDownLatch end = new CountDownLatch(PLAYER_AMOUNT);
Player[] plays = new Player[PLAYER_AMOUNT];

for(int i=0;i<PLAYER_AMOUNT;i++)
plays[i] = new Player(i+1,begin,end);

//设置特定的线程池,大小为5
ExecutorService exe = Executors.newFixedThreadPool(PLAYER_AMOUNT);
for(Player p:plays)
exe.execute(p); //分配线程
System.out.println("Race begins!");
begin.countDown();
try{
end.await(); //等待end状态变为0,即为比赛结束
}catch (InterruptedException e) {
// TODO: handle exception
e.printStackTrace();
}finally{
System.out.println("Race ends!");
}
exe.shutdown();
}
}
public class Player implements Runnable {

private int id;
private CountDownLatch begin;
private CountDownLatch end;
public Player(int i, CountDownLatch begin, CountDownLatch end) {
// TODO Auto-generated constructor stub
super();
this.id = i;
this.begin = begin;
this.end = end;
}

@Override
public void run() {
// TODO Auto-generated method stub
try{
begin.await(); //等待begin的状态为0
Thread.sleep((long)(Math.random()*100)); //随机分配时间,即运动员完成时间
System.out.println("Play"+id+" arrived.");
}catch (InterruptedException e) {
// TODO: handle exception
e.printStackTrace();
}finally{
end.countDown(); //使end状态减1,最终减至0
}
}
}


exchanger类:

Exchanger可以在两个线程之间交换数据,只能是2个线程,他不支持更多的线程之间互换数据。

当线程A调用Exchange对象的exchange()方法后,他会陷入阻塞状态,直到线程B也调用了exchange()方法,然后以线程安全的方式交换数据,之后线程A和B继续运行

  1. public class ThreadLocalTest {  
  2.   
  3.     public static void main(String[] args) {  
  4.         Exchanger<List<Integer>> exchanger = new Exchanger<>();  
  5.         new Consumer(exchanger).start();  
  6.         new Producer(exchanger).start();  
  7.     }  
  8.   
  9. }  
  10.   
  11. class Producer extends Thread {  
  12.     List<Integer> list = new ArrayList<>();  
  13.     Exchanger<List<Integer>> exchanger = null;  
  14.     public Producer(Exchanger<List<Integer>> exchanger) {  
  15.         super();  
  16.         this.exchanger = exchanger;  
  17.     }  
  18.     @Override  
  19.     public void run() {  
  20.         Random rand = new Random();  
  21.         for(int i=0; i<10; i++) {  
  22.             list.clear();  
  23.             list.add(rand.nextInt(10000));  
  24.             list.add(rand.nextInt(10000));  
  25.             list.add(rand.nextInt(10000));  
  26.             list.add(rand.nextInt(10000));  
  27.             list.add(rand.nextInt(10000));  
  28.             try {  
  29.                 list = exchanger.exchange(list);  
  30.             } catch (InterruptedException e) {  
  31.                 // TODO Auto-generated catch block  
  32.                 e.printStackTrace();  
  33.             }  
  34.         }  
  35.     }  
  36. }  
  37.   
  38. class Consumer extends Thread {  
  39.     List<Integer> list = new ArrayList<>();  
  40.     Exchanger<List<Integer>> exchanger = null;  
  41.     public Consumer(Exchanger<List<Integer>> exchanger) {  
  42.         super();  
  43.         this.exchanger = exchanger;  
  44.     }  
  45.     @Override  
  46.     public void run() {  
  47.         for(int i=0; i<10; i++) {  
  48.             try {  
  49.                 list = exchanger.exchange(list);  
  50.             } catch (InterruptedException e) {  
  51.                 // TODO Auto-generated catch block  
  52.                 e.printStackTrace();  
  53.             }  
  54.             System.out.print(list.get(0)+", ");  
  55.             System.out.print(list.get(1)+", ");  
  56.             System.out.print(list.get(2)+", ");  
  57.             System.out.print(list.get(3)+", ");  
  58.             System.out.println(list.get(4)+", ");  
  59.         }  
  60.     }  
  61. }

多线程并发常用类:condition,semaphore,CyclicBarrier,countdownlatch,exchanger使用整理

标签:

原文地址:http://blog.csdn.net/feiyang123_/article/details/43190561

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!