使用多线程之间的通信模拟队列

深藏阁楼爱情的钟 2022-06-07 12:44 240阅读 0赞

在java多线程中，JDK为我们提供了wait和notify实现对进程的控制，两者要和synchronized关键字一起使用，并且需要先获取对象锁。wait是释放锁，notify是不释放锁。

举个例子：

ListDemo代码：

public class ListDemo{
        
        private static LinkedList<Integer> list = new LinkedList<>();
        
        public static void add(){
            list.add(1);
        }
        
        public static int size(){
            return list.size();
        }
    }

线程A：

public class ThreadA implements Runnable {
        private Object obj;
        ThreadA(){
    
        }
        public ThreadA(Object obj){
            this.obj = obj;
        }
    
        @Override
        public void run() {
            try{
                synchronized (obj){
                    if(MyList.size()!=5){
                        System.out.println("wait begin"+System.currentTimeMillis());
                        obj.wait();
                        System.out.println("wait end"+System.currentTimeMillis());
                    }
                }
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    }

线程B：

public class ThreadB implements Runnable{
        private Object obj;
        ThreadB(){
    
        }
        public ThreadB(Object obj){
            super();
            this.obj = obj;
        }
        @Override
        public void run() {
            try{
                synchronized (obj){
                    for(int i=0;i<10;i++){
                        MyList.add();
                        if(MyList.size()==5){
                            obj.notify();
                            System.out.println("已经发出通知");
                        }
                        System.out.println("添加了"+i+"个元素");
                        Thread.sleep(1000);
                    }
                }
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    }

主方法：

public class Main {
        public static void main(String[] args){
            try{
                Object obj = new Object();
                new Thread(new ThreadA(obj)).start();
                Thread.sleep(1000);
                new Thread(new ThreadB(obj)).start();
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    }

运行结果：

![20171021213940778][]  
线程A一直等到线程B执行结束后才继续执行，可见wait释放锁，而notify不释放锁。

### 使用wait和notify模拟队列： ###

使用wait和notify实现阻塞的放入和取出（先进先出），思路：初始为队列的最大长度为5；添加一个元素时，判断队列大小是否为5，若是则等待添加；移除一个元素是，判断队列大小是否为0，若是则等待移除。实现代码如下：

public class MyQueue {
        //用一个线程安全的List实现数据存放
        private final LinkedList<Object> list = new LinkedList<>();
        private final AtomicInteger count = new AtomicInteger(); //计数器
    
        private final int maxSize = 5;
        private final int minSize = 0;
    
        //初始化锁对象
        private final Object lock = new Object();
    
        /**
         * put方法
         */
        public void put(Object obj){
            synchronized (lock){
                while (count.get() == maxSize){
                    try{
                        lock.wait();
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
                list.add(obj);
                //在AtomicInteger中getAndIncrement以原子方式将当前值加1。
                count.getAndIncrement();
                System.out.println("元素"+obj+"被添加");
                lock.notify();
            }
        }
    
        /**
         * get方法
         */
        public Object get(){
            Object temp;
            synchronized (lock){
                while (count.get()==minSize){
                    try {
                        lock.wait();
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
                count.getAndDecrement();
                //获取List的第一个元素并且返回
                temp = list.removeFirst();
                System.out.println("元素"+temp+"被移除");
                lock.notify();
            }
            return  temp;
        }
    
        public static void main(String[] args) throws InterruptedException{
            final MyQueue myQueue = new MyQueue();
            initQueue(myQueue);
            Thread th1 = new Thread(new Runnable() {
                @Override
                public void run() {
                    myQueue.put("f");
                    myQueue.put("g");
                }
            });
    
            Thread th2 = new Thread(new Runnable() {
                @Override
                public void run() {
                    try{
                        Thread.sleep(1000);
                        myQueue.get();
                        Thread.sleep(1000);
                        myQueue.get();
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            });
            th1.start();
            Thread.sleep(1000);
            th2.start();
        }
    
        private int size(){
            return count.get();
        }
    
        private static void initQueue(MyQueue myQueue){
            myQueue.put("a");
            myQueue.put("b");
            myQueue.put("c");
            myQueue.put("d");
            myQueue.put("e");
            System.out.println("当前个数"+myQueue.size());
        }
    }

运行结果：

![20171021214838628][]

[20171021213940778]: /images/20220607/06319da43f7e4d37a5a8718691275c8a.png
[20171021214838628]: /images/20220607/48dfb0a2f2494b199bd356d34f78f1b5.png