Java：线程如何双向通知

缺乏、安全感 2024-03-22 19:29 24阅读 0赞

### 线程如何双向通知 ###

办法1：wait()与notify()机制。  
办法2：Condition机制。

#### wait()与notify()机制 ####

##### 单个生产者单个消费者线程的情形： #####

public class Main {
        
        public static void main(String[] args) {
        
            MyQueue myQueue = new MyQueue();
            ProducerThread producerThread = new ProducerThread(myQueue);
            ConsumerThread consumerThread = new ConsumerThread(myQueue);
            producerThread.start();
            consumerThread.start();
        }
    }

public class MyQueue {
        
        
        private String[] data = new String[10];
        private int getIndex = 0;
        private int putIndex = 0;
        private int size = 0;
        public synchronized void put(String element) {
        
            
            if (size == data.length) {
        
                try {
        
                    wait();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
            data[putIndex] = element;
            ++putIndex;
            if (putIndex == data.length) putIndex = 0;
            ++size;
            notify();
        }
        
        public synchronized String get() {
        
            if (size == 0) {
        
                try {
        
                    wait();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
            String result = data[getIndex];
            ++getIndex;
            if (getIndex == data.length) getIndex = 0;
            --size;
            notify();
            return result;
        }
    }

import java.util.Random;
    
    public class ProducerThread extends Thread {
        
        
        private final MyQueue myQueue;
        private final Random random = new Random();
        private int index = 0;
        public ProducerThread(MyQueue myQueue) {
        
            this.myQueue = myQueue;
        }
        
        @Override
        public void run() {
        
            while (true) {
        
                
                String tmp = "ele-" + index;
                myQueue.put(tmp);
                System.out.println("添加元素：" + tmp);
                index++;
                try {
        
                    Thread.sleep(random.nextInt(1000));
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        }
    }

import java.util.Random;
    
    public class ConsumerThread extends Thread {
        
        
        private final MyQueue myQueue;
        private final Random random = new Random();
        
        public ConsumerThread(MyQueue myQueue) {
        
            this.myQueue = myQueue;
        }
        
        @Override
        public void run() {
        
            while (true) {
        
                String s = myQueue.get();
                System.out.println("\t\t消费元素：" + s);
                try {
        
                    Thread.sleep(random.nextInt(1000));
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        }
    }

##### 多个生产者多个消费者的情形： #####

public class Main2 {
        
        
        public static void main(String[] args) {
        
            MyQueue2 myQueue = new MyQueue2();
            for (int i = 0; i < 3; i++) {
        
                new ConsumerThread(myQueue).start();
            }
            for (int i = 0; i < 5; i++) {
        
                new ProducerThread(myQueue).start();
            }
        }
    }

public class MyQueue2 extends MyQueue {
        
        
        private String[] data = new String[10];
        private int getIndex = 0;
        private int putIndex = 0;
        private int size = 0;
        
        @Override
        public synchronized void put(String element) {
        
            if (size == data.length) {
        
                try {
        
                    wait();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
                put(element);
            } else {
        
                put0(element);
                notify();
            }
        }
        
        private void put0(String element) {
        
            data[putIndex] = element;
            ++putIndex;
            if (putIndex == data.length) putIndex = 0;
            ++size;
        }
        
        @Override
        public synchronized String get() {
        
            if (size == 0) {
        
                try {
        
                    wait();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
                return get();
            } else {
        
                String result = get0();
                notify();
                return result;
            }
        }
        
        private String get0() {
        
            String result = data[getIndex];
            ++getIndex;
            if (getIndex == data.length) getIndex = 0;
            --size;
            return result;
        }
    }