等待/通知范式的生产者消费者 java实现

素颜马尾好姑娘i 2021-12-20 03:17 128阅读 0赞

# 等待/通知范式的单生产者单消费者模型 #

package Concurrency;
    
    public class OneProOneCon {
        //等待/通知写法的单生产者单消费者模型
        public static boolean full=false;
        public static void main(String[] args) {
            Object lock = new Object();
            Consumer  c = new Consumer(lock);
            Producer  p = new Producer(lock);
            ThreadC tc = new ThreadC(c);
            ThreadP tp = new ThreadP(p);
            tp.start();
            tc.start();
    
        }
    
        static class Consumer{
            private Object lock;
            private int count;
            Consumer(Object lock){
                this.lock=lock;
                this.count=0;
            }
    
            public void consume(){
                try {
                    synchronized (lock){
                        while(!full) {
                            lock.wait();
                        }
                        count++;
                        System.out.println("a consumer consume "+count);
                        full=false;
                        lock.notify();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        static class Producer{
            private Object lock;
            private int count;
            Producer(Object lock){
                this.lock =lock;
                this.count=0;
            }
    
            public void product(){
                try {
                    synchronized (lock){
                        while(full){
                            lock.wait();
                        }
                        count++;
                        System.out.println("producer product a thing");
                        full=true;
                        lock.notify();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        static class ThreadP extends Thread{
            private Producer producer;
            ThreadP(Producer producer){
                this.producer = producer;
            }
    
            @Override
            public void run() {
                while(true) {
                    producer.product();
                }
            }
        }
    
        static class ThreadC extends Thread{
            private Consumer consumer;
            ThreadC(Consumer consumer){
                this.consumer=consumer;
            }
    
            @Override
            public void run() {
                try {
                    for(int i=0;i<3;i++){
                        consumer.consume();
                        Thread.sleep(2000);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
            }
        }
    
    
    }

# 等待/通知范式的多生产者多消费者实现 #

package Concurrency;
    
    public class MulProMulCon {
        //等待/通知模式的多生产者多消费者
        public static final int MAXPRODUCTNUM=5;
        public static int productLeft=0;
        public static void main(String[] args) {
            Object lock = new Object();
            ThreadC[] tcs = new ThreadC[10];
            ThreadP[] tps = new ThreadP[10];
            for(int i=0;i<10;i++){
                Consumer  c = new Consumer(lock,"consumer"+i);
                Producer  p = new Producer(lock,"producer"+i);
                tcs[i] = new ThreadC(c);
                tps[i] = new ThreadP(p);
            }
            for(int i=0;i<10;i++){
                tcs[i].start();
            }
            for(int i=0;i<10;i++){
                tps[i].start();
            }
        }
    
        static class Consumer{
            private String name;
            private Object lock;
            private int count;
            Consumer(Object lock,String name){
                this.lock=lock;
                this.count=0;
                this.name=name;
            }
    
            public void consume(){
                try {
                    synchronized (lock){
                        while(productLeft==0) {
                            lock.wait();
                        }
                        count++;
                        System.out.println(name+" consume a thing");
                        productLeft--;
                        lock.notifyAll();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        static class Producer{
            private Object lock;
            private String name;
            private int count;
            Producer(Object lock,String name){
                this.lock =lock;
                this.count=0;
                this.name=name;
            }
    
            public void product(){
                try {
                    synchronized (lock){
                        while(productLeft==MAXPRODUCTNUM){
                            lock.wait();
                        }
                        count++;
                        System.out.println(name+" product a thing");
                        productLeft++;
                        lock.notifyAll();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        static class ThreadP extends Thread{
            private Producer producer;
            ThreadP(Producer producer){
                this.producer = producer;
            }
    
            @Override
            public void run() {
                while(true) {
                    producer.product();
                }
            }
        }
    
        static class ThreadC extends Thread{
            private Consumer consumer;
            ThreadC(Consumer consumer){
                this.consumer=consumer;
            }
    
            @Override
            public void run() {
                try {
                    while(true){
                        consumer.consume();
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
            }
        }
    }

转载于:https://www.cnblogs.com/zhouyu0-0/p/10677388.html