java并发编程实战学习（3）--基础构建模块

谁践踏了优雅 2022-07-16 07:29 189阅读 0赞

转自：java并发编程实战

## 5.3阻塞队列和生产者-消费者模式 ##

BlockingQueue阻塞队列提供可阻塞的put和take方法，以及支持定时的offer和poll方法。如果队列已经满了，那么put方法将阻塞直到空间可用；如果队列为空，那么take方法将阻塞直到有元素可用。队列可以是有界的也可以是无界的。

如果生产者生成工作的速率比消费者处理工作的速率款，那么工作项会在队列中累计起来，最终好紧内存。同样，put方法的阻塞特性也极大地简化了生产者的编码。如果使用有界队列，当队列充满时，生产者将阻塞并不能继续生产工作，而消费者就有时间来赶上工作的进度。阻塞队列同样提供了一个offer方法，如果数据项不能被添加到队列中，那么将返回一个失败的状态。这样你就能创建更多灵活的策略来处理负荷过载的情况。

> 在构建高可靠的应用程序时，有界队列是一种强大的资源管理工具：他们能一直并防止产生过多的工作项，使应用程序在负荷过载的情况下边的更加健壮。

/**
     * java并发编程实战
     * 5.3.1桌面搜索
     * 爬虫查找所有文件并放入队列
     * Created by mrf on 2016/3/7.
     */
    public class FileCrawler implements Runnable {
        private final BlockingQueue<File> fileQueue;
        private final FileFilter fileFilter;
        private final File root;
    
        public FileCrawler(BlockingQueue<File> fileQueue, FileFilter fileFilter, File root) {
            this.fileQueue = fileQueue;
            this.fileFilter = fileFilter;
            this.root = root;
        }
    
        @Override
        public void run() {
            try {
                crawl(root);
            } catch (InterruptedException e) {
                //恢复中断
                Thread.currentThread().interrupt();
                e.printStackTrace();
            }
        }
    
        private void crawl(File root) throws  InterruptedException {
            File[] entries = root.listFiles(fileFilter);
            if (entries!=null){
                for (File entry : entries) {
                    if (entry.isDirectory()){
                        crawl(entry);
                    }else if (!alreadyIndexed(entry)){
                        fileQueue.put(entry);
                    }
                }
            }
        }
    
        private boolean alreadyIndexed(File entry){
            //检查是否加入索引
            return false;
        }
    }
    
    /**
     * 消费者
     * 将爬虫结果队列取出并加入索引
     */
    class Indexer implements Runnable{
        private static final int BOUND = 100;
        private static final int N_CONSUMERS = 2;
        private final BlockingQueue<File> queue;
    
        Indexer(BlockingQueue<File> queue) {
            this.queue = queue;
        }
    
        @Override
        public void run() {
            try {
                while (true){
                    indexFile(queue.take());
                }
            }catch (InterruptedException e){
                Thread.currentThread().interrupt();
            }
        }
    
        private void indexFile(File take) {
            //将文件加入索引
        }
    
        public static void startIndexing(File[] roots){
            BlockingQueue<File> queue = new LinkedBlockingDeque<>(BOUND);
            FileFilter fileFilter = new FileFilter() {
                @Override
                public boolean accept(File pathname) {
                    return true;
                }
            };
            for (File root:roots) {
                new Thread(new FileCrawler(queue,fileFilter,root)).start();
            }
            for (int i = 0; i < N_CONSUMERS; i++) {
                new Thread(new Indexer(queue)).start();
            }
        }
    }

## 　　5.5信号量 ##

Semaphore中管理着一组虚拟的许可（permit）。许可的初始数量可通过构造函数来指定。在执行操作时可以首先获得许可（只要还有剩余的许可），并在使用以后释放许可。如果没有许可，那么acquire将阻塞直到有许可（或者被中断或者操作超时）。release方法将返回一个许可给信号量。计算信号量的一种简化形式是二值信号量，即初始值为1的Semaphore。二值信号量可以用作互斥体（mutex），并具备不可重入的加锁语义：谁拥有这个唯一的许可，谁就拥有了互斥锁。

/**
     * java 并发编程实战
     * 5-14使用Semaphore做容器设置边界
     * 信号量
     * Created by mrf on 2016/3/8.
     */
    public class BoundedHashSet<T> {
        private final Set<T> set;
        private final Semaphore sem;
    
    //    public BoundedHashSet(Set<T> set, Semaphore sem) {
    //        this.set = set;
    //        this.sem = 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;
        }
    }

## 　　5.6构建高效且可伸缩的结果缓存 ##

/**
     * java并发编程实战
     * 5-16使用HashMap和不同机制来初始化缓存
     * 实现将曾经计算过的命令缓存起来，方便相同的计算直接出结果而不用重复计算
     * Created by mrf on 2016/3/8.
     */
    public interface Computable<A,V> {
        V compute(A arg) throws InterruptedException;
    }
    
    class ExpensiveFunction implements Computable<String,BigInteger>{
    
        @Override
        public BigInteger compute(String arg) throws InterruptedException {
            //在经过长时间的计算后
            return new BigInteger(arg);
        }
    }
    
    /**
     * 保守上锁办法
     * 每次只有一个线程能执行compute，性能差
     * @param <A>
     * @param <V>
     */
    class Memoizer1<A,V> implements Computable<A,V>{
        @GuardedBy("this")
        private final Map<A,V> cache = new HashMap<>();
        private final Computable<A,V> c;
    
        public Memoizer1(Computable<A, V> c) {
            this.c = c;
        }
    
        @Override
        public synchronized  V compute(A arg) throws InterruptedException {
            V result = cache.get(arg);
            if (result==null){
                result = c.compute(arg);
                cache.put(arg,result);
            }
            return result;
        }
    }
    
    /**
     * 5-17
     * 改用ConcurrentHashMap增强并发性
     * 但还有个问题，就是只有计算完的结果才能缓存，正在计算的没有缓存，
     * 这将导致一个长时间的计算没有放入缓存，另一个又开始重复计算。
     * @param <A>
     * @param <V>
     */
    class Memoizer2<A,V> implements Computable<A,V>{
    
        private final  Map<A,V> cache = new ConcurrentHashMap<>();
        private final  Computable<A,V> c;
    
        Memoizer2(Computable<A, V> c) {
            this.c = c;
        }
    
        @Override
        public V compute(A arg) throws InterruptedException {
            V result = cache.get(arg);
            if (result ==null){
                result = c.compute(arg);
                cache.put(arg,result);
            }
            return result;
        }
    }
    
    /**
     * 几乎完美：非常好的并发性，缓存正在计算中的结果
     * 但compute模块中if代码块是非原子性的，这样可能导致两个相同的计算
     * @param <A>
     * @param <V>
     */
    class Memoizer3<A,V> implements Computable<A,V>{
        private final Map<A,Future<V>> cache = new ConcurrentHashMap<>();
        private final Computable<A,V> c;
    
        Memoizer3(Computable<A, V> c) {
            this.c = c;
        }
    
        @Override
        public V compute(final A arg) throws InterruptedException {
            Future<V> f = cache.get(arg);
            if (f==null){
                Callable<V> eval = new Callable<V>() {
                    @Override
                    public V call() throws Exception {
                        return c.compute(arg);
                    }
                };
                FutureTask<V> ft = new FutureTask<V>(eval);
                f = ft;
                cache.put(arg,ft);
                ft.run();
            }
            try {
                return f.get();
            } catch (ExecutionException e) {
                //抛出正在计算
                e.printStackTrace();
            }
            return null;
        }
    }
    
    /**
     * 使用ConcurrentHashMap的putIfAbsent解决原子问题
     * 若计算取消则移除
     * @param <A>
     * @param <V>
     */
    class Memoizer<A,V> implements Computable<A,V>{
        private final ConcurrentHashMap<A,Future<V>> cache = new ConcurrentHashMap<>();
        private final Computable<A,V> c;
    
        Memoizer(Computable<A, V> c) {
            this.c = c;
        }
    
        @Override
        public V compute(final A arg) throws InterruptedException {
            while (true){
                Future<V> f = cache.get(arg);
                if (f==null){
                    Callable<V> eval = new Callable<V>() {
                        @Override
                        public V call() throws Exception {
                            return c.compute(arg);
                        }
                    };
                    FutureTask<V> ft = new FutureTask<V>(eval);
                    f = cache.putIfAbsent(arg,ft);
                    if (f==null){
                        f = ft;ft.run();
                    }
                }
                try {
                    return f.get();
                } catch (CancellationException e){
                    cache.remove(arg,f);
                } catch(ExecutionException e) {
                    //抛出正在计算
                    e.printStackTrace();
                }
                return null;
            }
    
        }
    }

## 　　小结： ##

*  可变状态是直观重要的（It's the mutable state,stupid）。所有的并发问题都可以归结为如何协调对并发状态的访问。可变状态越少，就越容易确保线程的安全性。
 *  尽量将域声明为final类型，除非需要他们是可变的。
 *  不可变对象一定是线程安全的。不可变对象能极大地降低并发编程的复杂性。他们更为简单而且可以任意共享而无须使用加锁或保护性复制等机制。
 *  封装有助于管理复杂性。在编写线程安全的程序时，虽然可以将所有数据都保存在全局变量中，但为什么要这样做？将数据封装在对象中，更易于维持不变性条件：将同步机制封装在对象中，更易于遵循同步策略。
 *  用锁来保护每个可变变量。
 *  当保护同一个不变性条件中的所有变量时，要使用同一个锁。
 *  在执行复合操作期间，要持有锁。
 *  如果从多个线程中访问同一个可变变量时没有同步机制，那么程序会出现问题。
 *  不要故作聪明地腿短出不需要使用同步。
 *  在设计过程中考虑线程安全，或者在文档中明确地指出他不是线程安全的。
 *  将同步策略文档化。