Java多线程实现生产者消费者模型

Posted 2020-12-12 optor

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

class Solution {

    static class Queue<T> {

        private int capacity;
        private Object[] elementData;

        private int size;
        private int consumePoint;
        private int productPoint;

        private Lock lock;
        private Condition emptyCondition;
        private Condition fullCondition;

        public Queue(int capacity) {
            this.capacity = capacity;
            elementData = new Object[capacity];

            size = 0;
            consumePoint = 0;
            productPoint = 0;

            lock = new ReentrantLock();
            emptyCondition = lock.newCondition();
            fullCondition = lock.newCondition();
        }

        @SuppressWarnings("unchecked")
        public T get() {
            Object data;
            lock.lock();
            try {
                while (size == 0) {
                    try {
                        emptyCondition.await();
                    } catch (InterruptedException ignored) {
                    }
                }

                if (consumePoint == capacity) {
                    consumePoint = 0;
                }
                data = elementData[consumePoint++];
                size--;
                fullCondition.signalAll();
            } finally {
                lock.unlock();
            }
            return (T) data;
        }

        public void put(T data) {
            lock.lock();
            try {
                while (size == capacity) {
                    try {
                        fullCondition.await();
                    } catch (InterruptedException ignored) {
                    }
                }

                if (productPoint == capacity) {
                    productPoint = 0;
                }
                elementData[productPoint++] = data;
                size++;
                emptyCondition.signalAll();
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException, IOException {
        Queue<Integer> queue = new Queue<>(5);

        int consumeThreadCount = 5, productThreadCount = 5;
        List<Thread> consumeThreads = new ArrayList<>(consumeThreadCount);
        List<Thread> productThreads = new ArrayList<>(productThreadCount);

        for (int i = 0; i < 5; i++) {
            Thread consumeThread = new Thread(() -> {
                while (true) {
                    Integer num = queue.get();
                    System.out.println(String.format("[%s] GET %d", Thread.currentThread().getName(), num));
                }
            });
            consumeThread.setName(String.format("consumeThread-%d", i));
            consumeThreads.add(consumeThread);

            Thread productThread = new Thread(() -> {
                while (true) {
                    int num = ThreadLocalRandom.current().nextInt(100);
                    queue.put(num);
                    System.out.println(String.format("[%s] PUT %d", Thread.currentThread().getName(), num));
                }
            });
            productThread.setName(String.format("productThread-%d", i));
            productThreads.add(productThread);
        }

        for (int i = 0; i < 5; i++) {
            consumeThreads.get(i).start();
            productThreads.get(i).start();
        }

        System.in.read();

    }
}