用wait/notify 模拟Queue

水深无声 2023-08-17 17:43 106阅读 0赞

一、BlockingQueue支持阻塞队列  
    此例子主要是模拟LinkBlockingQueue中put和take方法

public class SimulationQueue { 
        // 1 装载元素的容器
        private final List<Object> list = Lists.newArrayList();
        // 2 计数器
        private final AtomicInteger count = new AtomicInteger(0);
        // 3 确定容器的上限和下限
        private Integer maxSize;
        private Integer minSize;
    
        // 4 构造初始值
        public SimulationQueue(int maxSize) { 
            this.maxSize = maxSize;
        }
    
        // 5 初始化一个对象,用于加锁
        private Object lock = new Object();
    
    
        /** * put(an object) 把an object加到BlockingQueue里面，如果BlockingQueue没有空间 * 调用此方法的线程被阻断，直到BlockingQueue，有空间添加数据 * * @param obj */
        public void put(Object obj) { 
            synchronized (lock) { 
                while (count.get() == maxSize) { 
                    try { 
                        lock.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                // 1 容器中添加元素
                list.add(obj);
                // 2 计数器加1
                count.incrementAndGet();
                System.out.println("新加入的元素为：" + obj);
                // 3 通知其他线程(唤醒）
                lock.notify();
            }
        }
    
        /** * take() 取走排在BlockingQueue里面排在首位的元素，如果BlockingQueue为空 * 阻断进入等待状态，直到BlockingQueue有新的数据加入 */
        public Object take() { 
            Object ret = null;
            synchronized (lock) { 
                while (count.get() == 0) { 
                    try { 
                        lock.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                // 1 容器中取出排在首位的元素
                Object firstObj = list.remove(0);
                // 2 计数器递减
                count.decrementAndGet();
                // 3 通知其他线程（唤醒)
                lock.notify();
            }
            return ret;
        }
    
        public int size() { 
            return count.get();
        }
    
        public static void main(String[] args) throws  Exception { 
    
            final SimulationQueue m = new SimulationQueue(5);
            m.put("a");
            m.put("b");
            m.put("c");
            m.put("d");
            m.put("e");
            System.out.println("当前元素个数：" + m.size());
            Thread t1 = new Thread(new Runnable() { 
                @Override
                public void run() { 
                    m.put("h");
                    m.put("i");
                }
            }, "t1");
    
            Thread t2 = new Thread(new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        Thread.sleep(1000);
                        Object t1 = m.take();
                        //System.out.println("被取走的元素为：" + t1);
                        Thread.sleep(1000);
                        Object t2 = m.take();
                        //System.out.println("被取走的元素为：" + t2);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }, "t2");
    
            t1.start();
            Thread.sleep(1000);
            t2.start();
    
        }
    }