Java并发之同步工具类

2017-05-10  本文已影响60人  Showdy

同步工具类

闭锁(CountDownLatch)

CountDownLatch 方法解析

CountDownLatch可以使一个或者多个线程等待一组事件的发生.闭锁状态包括一个计数器,该计数器初始化为一个正数,表示需要等待的事件数量.countDown()表示递减计数器,表示有个一个事件发生,而await()等待计数器达到零,这表示所有等待的事件都已经发生.如果计数器值非0,那么await()则会阻塞直到计数器为0,或者线程中断,或者等待超时.

测试例子

创建一定数量的线程,利用他们并发执行指定的任务,计算出总共花了多少时间.

    
    public class TestHarness {
        public long 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();//任务执行完毕,计数器减1;
                            }
                        } catch (InterruptedException ignored) {
                        }
                    }
                };
                t.start();
            }
    
            long start = System.nanoTime();
            startGate.countDown(); //到此处,说明所有线程准备就绪,可以开始执行任务
            endGate.await();// 阻塞等待所有线程执行完毕.
            long end = System.nanoTime();
            return end - start;
        }
    }

栅栏(Barrier)

栅栏类似闭锁,能阻塞一组线程直到某个事件发生.栅栏与闭锁的区别关键在于:闭锁用于等待事件,而栅栏用于等待其他线程.

CyclicBarrier可以使得一定数量的参与反复在栅栏位置汇聚,在并行迭代算法中非常有用:将一个问题拆分成一系列相互独立的子问题.当线程达到栅栏位置时调用await()阻塞直到所有线程到达栅栏位置,如果所有线程到达,栅栏打开,线程唤醒,而栅栏被重置下次使用.如果await()调用超时,或者阻塞的线程被中断,那么栅栏就算是被打破,所有await()调用会抛出BrokenBrrierException.

CyclicBarrier还可以在构造函数中传入一个Runnable,当成功通过栅栏时会(在一个子线程中)执行他,但是阻塞线程被释放之前是不能执行的.

构造函数

循环栅栏函数解析
实例展示:

    public class CellularAutomata {
        private final Board mainBoard;
        private final CyclicBarrier barrier;
        private final Worker[] workers;
    
        public CellularAutomata(Board board) {
            this.mainBoard = board;
            int count = Runtime.getRuntime().availableProcessors();
            this.barrier = new CyclicBarrier(count,
                    new Runnable() {
                        public void run() {
                            mainBoard.commitNewValues();//栅栏打开,计算值
                        }});
            this.workers = new Worker[count];
            for (int i = 0; i < count; i++)
                workers[i] = new Worker(mainBoard.getSubBoard(count, i));
        }
    
        private class Worker implements Runnable {
            private final Board board;
    
            public Worker(Board board) { this.board = board; }
            public void run() {
                while (!board.hasConverged()) {
                    for (int x = 0; x < board.getMaxX(); x++)
                        for (int y = 0; y < board.getMaxY(); y++)
                            board.setNewValue(x, y, computeValue(x, y));
                    try {
                        barrier.await();//线程计算完毕等待其他线程
                    } catch (InterruptedException ex) {
                        return;
                    } catch (BrokenBarrierException ex) {
                        return;
                    }
                }
            }
    
            private int computeValue(int x, int y) {
                // Compute the new value that goes in (x,y)
                return 0;
            }
        }
    
        public void start() {
            for (int i = 0; i < workers.length; i++)
                new Thread(workers[i]).start();
            mainBoard.waitForConvergence();
        }
    
        interface Board {
            int getMaxX();
            int getMaxY();
            int getValue(int x, int y);
            int setNewValue(int x, int y, int value);
            void commitNewValues();
            boolean hasConverged();
            void waitForConvergence();
            Board getSubBoard(int numPartitions, int index);
        }
    }

信号量(Semaphore)

计数信号量(Counting Semaphore)用来控制同时访问某个特定资源的操作数量,或者执行某个指定操作的数量.还可以用来实现某种连接池,或者对容器加边界.

Semaphore管理着一组虚拟的许可(permit),许可的初始数量可通过构造函数指定,在执行操作时先要获得许可,并在使用后释放许可.如果没有许可,accquire将阻塞到有许可为止(或者被中断,或者操作超时),release()将返回一个许可信号量.

Semaphore类只是一个资源数量的抽象表示,并不负责管理资源对象本身,可能有多个线程同时获取到资源使用许可,因此需要使用同步机制避免数据竞争.

Semaphore函数解析

多线程同时操作特定资源例子

    public class SemaphoreDemo {
      
        private final ReentrantLock lock = new ReentrantLock();
        private final Semaphore semaphore;
        private final LinkedList<Object> resourceList = new LinkedList<Object>();
        private static CountDownLatch mCountDownLatch = new CountDownLatch(9);
    
        public SemaphoreDemo(Collection<Object> resourceList) {
            this.resourceList.addAll(resourceList);
            //公平模式
            this.semaphore = new Semaphore(resourceList.size(), true);
        }
    
       
        public Object acquire() throws InterruptedException {
            semaphore.acquire();
    
            lock.lock();
            try {
                return resourceList.pollFirst();
            } finally {
                lock.unlock();
            }
        }
    
        public void release(Object resource) {
            lock.lock();
            try {
                resourceList.addLast(resource);
            } finally {
                lock.unlock();
            }
    
            semaphore.release();
        }
    
        public static void main(String[] args) {
            //准备2个可用资源
            List<Object> resourceList = new ArrayList<>();
            resourceList.add("Resource1");
            resourceList.add("Resource2");
    
            //准备工作任务
            final SemaphoreDemo demo = new SemaphoreDemo(resourceList);
            Runnable worker = new Runnable() {
                @Override
                public void run() {
                    Object resource = null;
                    try {
                        //获取资源
                        resource = demo.acquire();
                        System.out.println(Thread.currentThread().getName() + "\twork   on\t" + resource);
                        //用resource做工作
                        Thread.sleep(1000);
                        System.out.println(Thread.currentThread().getName() + "\tfinish on\t" + resource);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        //归还资源
                        if (resource != null) {
                            demo.release(resource);
                            mCountDownLatch.countDown();
    
                        }
                    }
                }
            };
    
            //启动9个任务
            ExecutorService service = Executors.newCachedThreadPool();
            for (int i = 0; i < 9; i++) {
                service.submit(worker);
            }
    
    
            try {
                mCountDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            service.shutdown();
        }
    }
将任何一种容器变成有界阻塞容器

    public class BoundedHashSet <T> {
        private final Set<T> set;
        private final Semaphore sem;
    
        public BoundedHashSet(int bound) {
            this.set = Collections.synchronizedSet(new HashSet<T>());
            sem = new Semaphore(bound);
        }
    
        public boolean add(T o) throws InterruptedException {
            sem.acquire();
            boolean wasAdded = false;
            try {
                wasAdded = set.add(o);
                return wasAdded;
            } finally {
                if (!wasAdded)
                    sem.release();
            }
        }
    
        public boolean remove(Object o) {
            boolean wasRemoved = set.remove(o);
            if (wasRemoved)
                sem.release();
            return wasRemoved;
        }
    }
上一篇 下一篇

猜你喜欢

热点阅读