0
点赞
收藏
分享

微信扫一扫

生产者消费者模式的3种方法

扒皮狼 2022-01-31 阅读 74
java生产者消费者阻塞队列Java编程语言

该篇主要讲解生产者消费者模式的3种实现方式

生产者消费者模式

    我们先来看看什么是生产者消费者模式，生产者消费者模式是程序设计中非常常见的
一种设计模式，被广泛运用在解耦、消息队列等场景。在现实世界中，我们把生产商品的一
方称为生产者，把消费商品的一方称为消费者，有时生产者的生产速度特别快，但消费者的
消费速度跟不上，俗称“产能过剩”，又或是多个生产者对应多个消费者时，大家可能会手
忙脚乱。如何才能让大家更好地配合呢？这时在生产者和消费者之间就需要一个中介来进行
调度，于是便诞生了生产者消费者模式。

图片.png

使用生产者消费者模式通常需要在两者之间增加一个阻塞队列作为媒介，有了媒介之后就
相当于有了一个缓冲，平衡了两者的能力，整体的设计如图所示，最上面是阻塞队列，右
侧的 1 是生产者线程，生产者在生产数据后将数据存放在阻塞队列中，左侧的 2 是消费
者线程，消费者获取阻塞队列中的数据。而中间的 3 和 4 分别代表生产者消费者之间互
相通信的过程，因为无论阻塞队列是满还是空都可能会产生阻塞，阻塞之后就需要在合适
的时机去唤醒被阻塞的线程。

那么什么时候阻塞线程需要被唤醒呢？有两种情况。第一种情况是当消费者看到阻塞队列
为空时，开始进入等待，这时生产者一旦往队列中放入数据，就会通知所有的消费者，唤
醒阻塞的消费者线程。另一种情况是如果生产者发现队列已经满了，也会被阻塞，而一旦
消费者获取数据之后就相当于队列空了一个位置，这时消费者就会通知所有正在阻塞的生
产者进行生产，这便是对生产者消费者模式的简单介绍。

synchronized wait/notify 实现生产者消费者模式

MyBlocking.java

public class MyBlocking<T> {
    private int maxSize;
    private LinkedList<T> list;

    public MyBlocking(int maxSize) {
        this.maxSize = maxSize;
        list = new LinkedList();
    }

    public synchronized void put(T t) throws InterruptedException {
        while (list.size() == maxSize) {
            wait();
        }
        list.add(t);
        notifyAll();
    }

    public synchronized T take() throws InterruptedException {
        while (list.size() == 0) {
            wait();
        }
        T t =  list.remove();
        notifyAll();
        return t;
    }
}

生产者producer.java

public class Producer implements Runnable {

    private MyBlocking<Integer> queue;
    private Random random = new Random();
    public Producer(MyBlocking<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            try {
                int i = random.nextInt(1000);
                queue.put(i);
                System.out.println("producer:" + i + "-" + Thread.currentThread().getName());
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

消费者Consumer.java

public class Consumer implements Runnable {
    private MyBlocking<Integer> queue;
    public Consumer(MyBlocking<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while(true) {
            try {
                System.out.println("consume:" + queue.take());
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

实现

public class PC {
    public static void main(String[] args) {
        MyBlocking<Integer> queue = new MyBlocking<>(10);
        new Thread(new Producer(queue)).start();
        new Thread(new Producer(queue)).start();
        new Thread(new Producer(queue)).start();
        new Thread(new Consumer(queue)).start();
        new Thread(new Consumer(queue)).start();
        new Thread(new Consumer(queue)).start();
    }
}

Lock Condition signalAll await方式

MyBlockingForCondition.java

public class MyBlockingForCondition {
    private Queue queue;
    private int max = 16;
    private ReentrantLock lock = new ReentrantLock();
    private Condition notEmpty = lock.newCondition();
    private Condition notFull = lock.newCondition();

    public MyBlockingForCondition(int size) {
        this.max = size;
        queue = new LinkedList();
    }

    public void put(Integer value) throws InterruptedException {
        try {
            lock.lock();
            while (queue.size() == max) {
                notFull.await();
            }
            queue.add(value);
            notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public Integer take() throws InterruptedException {
        try {
            lock.lock();
            while (queue.size() == 0) {
                notEmpty.await();
            }
            Integer value = (Integer) queue.remove();
            notFull.signalAll();
            return value;

        } finally {
            lock.unlock();
        }
    }
}

生产者Producer

public class Producer implements Runnable {
    private MyBlockingForCondition queue;
    // 本身就是原子性，可以换成Random
    private AtomicInteger atomic = new AtomicInteger(1);

    public Producer(MyBlockingForCondition queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            try {
                int i = atomic.getAndIncrement();
                queue.put(i);
                System.out.println("producer:" + i + "-" + Thread.currentThread().getName());
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

消费者

public class Consumer implements Runnable {

    private MyBlockingForCondition queue;

    public Consumer(MyBlockingForCondition queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while(true) {
            try {
                System.out.println("consume:" + queue.take());
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

实现

public class PC {
    public static void main(String[] args) {
        MyBlockingForCondition queue = new MyBlockingForCondition(10);
        new Thread(new Producer(queue)).start();
        new Thread(new Producer(queue)).start();
        new Thread(new Producer(queue)).start();
        new Thread(new Consumer(queue)).start();
        new Thread(new Consumer(queue)).start();
        new Thread(new Consumer(queue)).start();
    }
}

BlockingQueue方式实现生产者消费者

生产者Producer

public class Producer implements Runnable {

    private BlockingQueue<Integer> queue;
    private Random random = new Random();
    public Producer(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            try {
                int i = random.nextInt(1000);
                queue.put(i);
                System.out.println("producer:" + i + "-" + Thread.currentThread().getName());
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

消费者

public class Consumer implements Runnable {
    private BlockingQueue<Integer> queue;

    public Consumer(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while(true) {
            try {
                System.out.println("consume:" + queue.take());
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

实现

public class PC {
    public static void main(String[] args) {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(10);
        new Thread(new Producer(queue)).start();
        new Thread(new Producer(queue)).start();
        new Thread(new Producer(queue)).start();
        new Thread(new Consumer(queue)).start();
        new Thread(new Consumer(queue)).start();
        new Thread(new Consumer(queue)).start();
    }
}

ArrayBlockingQueue源码

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    public boolean offer(E e) {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

    public void put(E e) throws InterruptedException {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

      public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }
    private void enqueue(E e) {
        // assert lock.isHeldByCurrentThread();
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[putIndex] = e;
        if (++putIndex == items.length) putIndex = 0;
        count++;
        notEmpty.signal();
    }

    /**
     * Extracts element at current take position, advances, and signals.
     * Call only when holding lock.
     */
    private E dequeue() {
        // assert lock.isHeldByCurrentThread();
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E e = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length) takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        notFull.signal();
        return e;
    }
}
举报

相关推荐

生产者消费者模式

技术分享

生产者消费者

flutterandroidandroid studio

异步模式之生产者/消费者

javaintellij-idea并发数据消息队列Linux系统/运维

异步模式之生产者消费者

javarabbitmq开发语言

生产者与消费者设计模式

ide等待状态可执行编程语言

RabbitMq消费者和生产者模式

中间件rabbitmq后端

设计模式之生产者/消费者模式

机器学习决策树人工智能

生产者与消费者

java开发语言后端

生产者消费者问题

golangc++

生产者消费者模型

linux
0 条评论