简单易懂的队列消费

编写queue

public class MyQueue {

    String[] data = new String[3];
    private int putIndex = 0;
    private int getIndex = 0;
    private int size = 0;

    public synchronized void put(String element) {
        //当队列存满，需要等待消费者消费
        if (size == data.length) {
            try {
                System.out.println(Thread.currentThread().getName() + ":满了满了，等下！！！");
                wait(); //3
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            put(element);
            return;
        }
        put0(element);

    }

    private void put0(String element) {
        //给队列赋值
        data[putIndex] = element;
        //队列有数据，则唤醒get()的等待线程
        notify();
        System.out.println(Thread.currentThread().getName() + ":put唤醒！！！" + size);
        //每次插入消息后角标加一
        ++putIndex;
        //如果角标达到队列的最大长度，需要赋值为0，从头开始
        if (putIndex == data.length) {
            putIndex = 0;
        }
        //通过size记录每次数组队列在存满消息时等待
        ++size;
    }

    public synchronized String get() throws InterruptedException {
        //size=0则没有消息，需要等待
        if (size == 0) {
            try {
                wait();
                System.out.println(Thread.currentThread().getName() + ":消费没了，喀喀喀！！！");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return get();
        }
        return get0();
    }

    private String get0() throws InterruptedException {
        Thread.sleep(500);
        String result = data[getIndex];
        System.out.println(Thread.currentThread().getName() + ":get唤醒！！！" + size);
        notify();
        ++getIndex;
        if (getIndex == data.length) {
            getIndex = 0;
        }
        --size;
        return result;
    }
}

编写提供者，向队列存入消息

public class ProviderThread extends Thread {

    private MyQueue myQueue;

    public ProviderThread(MyQueue myQueue) {
        this.myQueue = myQueue;
    }

    @Override
    public void run() {
        int i = 1;
        while (true) {
            try {
                String tmp = "线程名：" + Thread.currentThread().getName() + ",提供者提工消息：" + i++;
                myQueue.put(tmp);
                System.out.println(tmp);
                Thread.sleep(300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

编写消费者，从队列读取消息

public class ConsumerThread extends Thread {

    private MyQueue myQueue;

    public ConsumerThread(MyQueue myQueue){
        this.myQueue = myQueue;
    }

    @Override
    public void run() {
        while (true){
            try {
                String result = myQueue.get();
                System.out.println("线程名：" + Thread.currentThread().getName() + ",消费者消费的消息：" + result);
                Thread.sleep(600);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

编写启动类

public class QueueMain {

    public static void main(String[] args) throws InterruptedException {
        //一个队列，每个线程向队列里面插入消息
        MyQueue myQueue = new MyQueue();
        for (int i = 0; i < 3 ; i++) {
            Thread.sleep(100);
            new ProviderThread(myQueue).start();
        }
        for (int i = 0; i < 1; i++) {
            new ConsumerThread(myQueue).start();
        }
//        Thread.sleep(10000);
//        System.exit(0);
    }

}

• 未加递归之前
  

• 添加递归之后