使用Semaphore或Condition把任何容器转化为有界的阻塞容器

超、凢脫俗 2022-04-05 10:51 130阅读 0赞

信号量被初始化为容器所期望容量的最大值。add操作在向底层容器中添加条目之前，需要先 获取一个许可。事实上，如果add操作没有能加入任何东西，它会立刻释放一个许可，同样，一个成功的remove操作释放一个许可，使更多的元素能够加入其中。

我们常用的阻塞队列LinkedBlockingQueue是使用Condition来进行限制的，原理差不多。

private final Condition notEmpty = takeLock.newCondition();

private final Condition notFull = putLock.newCondition();

使用信号量来约束容器：

import java.util.Collections;
    import java.util.HashSet;
    import java.util.Set;
    import java.util.concurrent.Semaphore;
    public class BoundHashSet<T> {
    	private Set<T> set;
    	private Semaphore sem;
    	public BoundHashSet(int num){
    		//得到一个经过同步的hashset
    		set = Collections.synchronizedSet(new HashSet<T>());
    		sem = new Semaphore(5);//设最多可放入的元素个数为5个
    	}
    	public boolean add(T e) throws InterruptedException{
    		sem.acquire();
    		boolean isadd = false;
    		try{
    			isadd = set.add(e);
    		}finally{
    			if(!isadd){
    				sem.release();
    			}
    		}
    		return isadd;
    	}
    	public boolean remove(T e){
    		boolean isremove = set.remove(e);
    		if(isremove){
    			sem.release();
    		}
    		return isremove;
    	}
    }

下面使用Condition进行实现

import java.util.Random;
    import java.util.concurrent.ConcurrentLinkedQueue;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * @author Administrator
     */
    public class BoundHashSet {
    	Lock lock = new ReentrantLock();
    	Condition notFull = lock.newCondition();
    	Condition notEmpty = lock.newCondition();
    	private ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<Integer>();
    	public static void main(String[] args) {
    		final BoundHashSet b = new BoundHashSet();
    		final Random rand = new Random();
    		new Thread(){
    			public void run(){
    				while(true){
    					Integer value = rand.nextInt(100);
    					b.offer(value);
    				}
    			}
    		}.start();
    		new Thread(){
    			public void run(){
    				while(true){
    					b.take();
    				}
    			}
    		}.start();
    	}
    	public void offer(Integer value){
    		lock.lock();
    		try {
    			while(queue.size() == 5){
    				notFull.await();
    			}
    			queue.offer(value);
    			System.out.println(Thread.currentThread().getName()+"#"+queue.size()+"#插入一个元素:"+value);
    			notEmpty.signal();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}finally{
    			lock.unlock();
    		}
    	}
    	public Integer take(){
    		lock.lock();
    		Integer value = null;
    		try{
    			while(queue.size() == 0){
    				try {
    					notEmpty.await();
    				} catch (InterruptedException e) {e.printStackTrace();}
    			}
    			value = queue.poll();
    			System.out.println(Thread.currentThread().getName()+"#"+queue.size()+"#获得一个元素:"+value);
    			notFull.signal();
    		}finally{
    			lock.unlock();
    		}
    		return value;
    	}
    /**
     *  Thread-0#1#插入一个元素:37
    	Thread-0#2#插入一个元素:71
    	Thread-0#3#插入一个元素:41
    	Thread-0#4#插入一个元素:22
    	Thread-0#5#插入一个元素:36
    	Thread-1#4#获得一个元素:37
    	Thread-1#3#获得一个元素:71
    	Thread-1#2#获得一个元素:41
    	Thread-1#1#获得一个元素:22
    	Thread-1#0#获得一个元素:36
    	Thread-0#1#插入一个元素:83
    	Thread-0#2#插入一个元素:61
    	Thread-0#3#插入一个元素:15
    	Thread-0#4#插入一个元素:99
    	Thread-0#5#插入一个元素:49
    	Thread-1#4#获得一个元素:83
     */
    }