java 并发包 Lock Condition实现阻塞队列

一时失言乱红尘 2022-04-17 04:49 165阅读 0赞

import java.util.LinkedList;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class LockConditionTest {
    
        public static void main(String[] args) {
    
            MyQueue<String> queue = new MyQueue<>(5);
            while(true){
                try {
                    Thread.sleep(0);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
                new Thread(()->{
                    String s = queue.get();
                    System.out.println("发送消息："+s);
                }).start();
    
                new Thread(()->{
                    String set = new Random().nextInt()*1000+"";
                    queue.put(set);
                    System.out.println("存储消息："+set);
                }).start();
    
                System.out.println("对列长度："+queue.getCount());
            }
    
        }
    
    
        static class MyQueue<OBJ>{
    
            private final int total;//最大长度
            private List<OBJ>  list; //对列
            private Lock lock = new ReentrantLock(); //锁
            private Condition notFull = lock.newCondition();//不满
            private Condition notEmpty = lock.newCondition(); //不为空
    
            public MyQueue(int total){
                this.total = total;
                list = new LinkedList<>();
            }
    
            public int getCount(){
                return this.list.size();
            }
    
            public void put(OBJ obj){
                lock.lock();
                try{
                    //如果满了,则阻塞
                    if(list.size() >= total){
                        try {
                            System.out.println("满了，等待发送");
                            this.notFull.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    list.add(obj);
    
                    //不为空唤醒
                    notEmpty.signal();
                }finally {
                    lock.unlock();
                }
            }
    
            public OBJ get(){
                lock.lock();
                try{
                    if(list.isEmpty()){
                        try {
                            System.out.println("没有信息了，等待注入");
                            notEmpty.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    OBJ obj = list.remove(0);
                    notFull.signal();
                    return obj;
                }finally {
                    lock.unlock();
                }
            }
        }
    }