Java多线程之常用类--闭包、栅栏、信号量、FutureTask

Java多线程之常用类--闭包、栅栏、信号量、FutureTask

 

 

闭锁

 

用途：可用于命令一组线程在同一个时刻开始执行某个任务，或者等待一组相关的操作结束。尤其适合计算并发执行某个任务的耗时。

 

public class CountDownLatchTest {  
  
    public void timeTasks(int nThreads, final Runnable task) throws InterruptedException{  
        final CountDownLatch startGate = new CountDownLatch(1);  
        final CountDownLatch endGate = new CountDownLatch(nThreads);  
          
        for(int i = 0; i < nThreads; i++){  
            Thread t = new Thread(){  
                public void run(){  
                    try{  
                        startGate.await();  
                        try{  
                            task.run();  
                        }finally{  
                            endGate.countDown();  
                        }  
                    }catch(InterruptedException ignored){  
                          
                    }  
                      
                }  
            };  
            t.start();  
        }  
          
        long start = System.nanoTime();  
        System.out.println("打开闭锁");  
        startGate.countDown();  
        endGate.await();  
        long end = System.nanoTime();  
        System.out.println("闭锁退出，共耗时" + (end-start));  
    }  
      
    public static void main(String[] args) throws InterruptedException{  
        CountDownLatchTest test = new CountDownLatchTest();  
        test.timeTasks(5, test.new RunnableTask());  
    }  
      
    class RunnableTask implements Runnable{  
  
        @Override  
        public void run() {  
            System.out.println("当前线程为：" + Thread.currentThread().getName());  
              
        }     
    } 

 

执行结果为：  
打开闭锁  
当前线程为：Thread-0  
当前线程为：Thread-3  
当前线程为：Thread-2  
当前线程为：Thread-4  
当前线程为：Thread-1  
闭锁退出，共耗时1109195  

 

 

 

栅栏

 

用途：用于阻塞一组线程直到某个事件发生。所有线程必须同时到达栅栏位置才能继续执行下一步操作，且能够被重置以达到重复利用。而闭锁式一次性对象，一旦进入终止状态，就不能被重置。

 

• 闭锁用于等待事件
• 栅栏用于等待线程

public class CyclicBarrierTest {  
    private final CyclicBarrier barrier;  
    private final Worker[] workers;  
  
    public CyclicBarrierTest(){  
        int count = Runtime.getRuntime().availableProcessors();  
        this.barrier = new CyclicBarrier(count,  
                new Runnable(){  
  
                    @Override  
                    public void run() {  
                        System.out.println("所有线程均到达栅栏位置，开始下一轮计算");  
                    }  
              
        });  
        this.workers = new Worker[count];  
        for(int i = 0; i< count;i++){  
            workers[i] = new Worker(i);  
        }  
    }  
    private class Worker implements Runnable{  
        int i;  
          
        public Worker(int i){  
            this.i = i;  
        }  
  
        @Override  
        public void run() {  
            for(int index = 1; index < 3;index++){  
                System.out.println("线程" + i + "第" + index + "次到达栅栏位置，等待其他线程到达");  
                try {  
                    //注意是await,而不是wait  
                    barrier.await();  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                    return;  
                } catch (BrokenBarrierException e) {  
                    e.printStackTrace();  
                    return;  
                }  
            }  
        }  
          
    }  
      
    public void start(){  
        for(int i=0;i<workers.length;i++){  
            new Thread(workers[i]).start();  
        }  
    }  
      
    public static void main(String[] args){  
        new CyclicBarrierTest().start();  
    }  
} 

 

执行结果为：  
线程0第1次到达栅栏位置，等待其他线程到达  
线程1第1次到达栅栏位置，等待其他线程到达  
线程2第1次到达栅栏位置，等待其他线程到达  
线程3第1次到达栅栏位置，等待其他线程到达  
所有线程均到达栅栏位置，开始下一轮计算  
线程3第2次到达栅栏位置，等待其他线程到达  
线程2第2次到达栅栏位置，等待其他线程到达  
线程0第2次到达栅栏位置，等待其他线程到达  
线程1第2次到达栅栏位置，等待其他线程到达  
所有线程均到达栅栏位置，开始下一轮计算 

 

 

 

信号量

 

用途：用来控制同时访问某个特定资源的操作数量，或者同时执行某个指定操作的数量。计数信号量可以用来实现某种资源池，或者对容器施加边界。

 

public class SemaphoreTest<T> {  
    private final Set<T> set;  
      
    private final Semaphore sema;  
      
    public SemaphoreTest(int bound){  
        this.set = Collections.synchronizedSet(new HashSet<T>());  
        this.sema = new Semaphore(bound);  
    }  
      
    public boolean add(T o) throws InterruptedException{  
        sema.acquire();  
        boolean wasAdded = false;  
        try{  
            wasAdded = set.add(o);  
            return wasAdded;  
        }finally{  
            if(!wasAdded){  
                sema.release();  
            }  
        }  
    }  
      
    public boolean remove(T o){  
        boolean wasRemoved = set.remove(o);  
        if(wasRemoved){  
            sema.release();  
        }  
        return wasRemoved;  
    }  
      
    public static void main(String[] args) throws InterruptedException{  
        int permits = 5;  
        int elements = permits + 1;  
        SemaphoreTest<Integer> test = new SemaphoreTest<Integer>(permits);  
        for(int i = 0;i < elements; i++){  
            test.add(i);  
        }  
    }  
} 

 

输出结果：
由于实际待添加的元素个数大于信号量所允许的数量，因此最后一次添加时，会一直阻塞。

 

 

 

FutureTask

 

       当一个计算的代价比较高，譬如比较耗时，或者耗资源，为了避免重复计算带来的浪费，当第一次计算后，通常会将结果缓存起来。比较常见的方式就是使用synchronized进行同步，但该方式带来的代价是被同步的代码只能被串行执行，如果有多个线程在排队对待计算结果，那么针对最后一个线程的计算时间可能比没有使用缓存的时间会更长。

第二种方式是采用ConcurrentHashMap，但对于耗时比较长的计算过程来说，该方式也存在一个漏洞。如果在第一个线程正在计算的过程中，第二个线程开始获取结果，会发现缓存里没有缓存结果，因此第二个线程又启动了同样的计算，这样就导致重复计算，违背了缓存的初衷。计算过程越长，则出现这种重复计算的几率就会越大。

 

       通过第二种方式的缺点分析，得知真正要缓存的应该是计算是否已被启动，而不是等待漫长的计算过程结束后，再缓存结果。一旦从缓存中得知某个计算过程已被其他线程启动，则当前线程不需要再重新启动计算，只需要阻塞等待计算结果的返回。FutureTask就是实现该功能的最佳选择。

 

public class Memoizer<A, V> implements Computable<A, V> {  
    private ConcurrentMap<A, Future<V>> cache = new ConcurrentHashMap<A, Future<V>>();  
    private final Computable<A, V> c;  
  
    public Memoizer(Computable<A, V> c) {  
        this.c = c;  
    }  
  
    @Override  
    public V compute(final A a) {  
        while (true) {  
            Future<V> f = cache.get(a);  
            if (null == f) {  
                Callable<V> eval = new Callable<V>() {  
  
                    @Override  
                    public V call() throws Exception {  
                        return c.compute(a);  
                    }  
  
                };  
                FutureTask<V> ft = new FutureTask<V>(eval);  
                f = cache.putIfAbsent(a, ft);  
                if (null == f) {  
                    f = ft;  
                    ft.run();  
                }  
            }  
            try {  
                return f.get();  
            } catch (InterruptedException e) {  
                e.printStackTrace();  
                cache.remove(a, f);  
            } catch (ExecutionException e) {  
                e.printStackTrace();  
            }  
        }  
    }  
  
    public static void main(String[] args) throws InterruptedException,  
            ExecutionException {  
        Computable<Integer, String> c = new Computable<Integer, String>() {  
  
            @Override  
            public String compute(Integer a) {  
                try {  
                    Thread.sleep(1000);  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
                String result = "由线程：" + Thread.currentThread().getName()  
                        + "计算得到" + a + "的结果";  
                return result;  
            }  
        };  
  
        Memoizer<Integer, String> memoizer = new Memoizer<Integer, String>(c);  
  
        ExecutorService executorService = Executors.newFixedThreadPool(3);  
  
        for (int i = 0; i < 3; i++) {  
            Task<Integer, String> task = new Task<Integer, String>(memoizer, 10);  
            String result = executorService.submit(task).get();  
            System.out.println(result);  
        }  
        executorService.shutdown();  
    }  
} 

 

结果如下：  
由线程：pool-1-thread-1计算得到10的结果  
由线程：pool-1-thread-1计算得到10的结果  
由线程：pool-1-thread-1计算得到10的结果  
当然如果仅仅只是采用FutureTask，仅仅只是减小了启动同一个计算过程的概率。当两个线程同时经过compute方法时，还是会出现重复启动同一个计算的情况，上面的例子通过结合ConcurrentHashMap 的putIfAbsent方法解决了这个问题