原创：转载需注明原创地址 https://www.cnblogs.com/fanerwei222/p/11871704.html

package concurrent.blockingqueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.atomic.AtomicInteger;

/**

 * 生产者

 */

public class Producer implements Runnable {

    /**

     * 计数器

     */

    private static AtomicInteger k = new AtomicInteger(0);

    BlockingQueue queue;

    public Producer(BlockingQueue queue) {

        this.queue = queue;

    }

    @Override

    public void run() {

        while (true) {

            try {

                /**

                 * 把生产出来的东西丢进队列中去

                 */

                queue.put(produce());

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

        }

    }

    /**

     * 生产东西, 然后返回出去

     * @return

     * @throws InterruptedException

     */

    private AtomicInteger produce() throws InterruptedException {

        k.addAndGet(1);

        System.out.println("生产者开始生产东西了... " + k);

        return k;

    }

}

package concurrent.blockingqueue;

import java.util.concurrent.BlockingQueue;

/**

 * 消费者

 */

public class Consumer implements Runnable {

    BlockingQueue queue;

    public Consumer(BlockingQueue queue){

        this.queue = queue;

    }

    @Override

    public void run() {

        while (true) {

            try {

                /**

                 * 延迟3秒再从队列中取出,

                 * 这样就可以看到生产队列生产满了之后就停止生产,

                 * 直到消费者消费了之后再继续生产

                 */

                Thread.sleep(3000);

                consume(queue.take());

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

        }

    }

    /**

     * 消费东西

     * @param o

     */

    void consume(Object o){

        System.out.println("消费者开始消费东西了....  " + o);

    }

}

package concurrent.blockingqueue;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

/**

 * main 测试类

 */

public class MainTest {

    public static void main(String[] args) {

        /**

         * 搞一个阻塞队列容量为3, 当超过容量就开始堵塞

         */

        BlockingQueue queue = new ArrayBlockingQueue(3);

        new Thread(new Producer(queue)).start();

        new Thread(new Consumer(queue)).start();

    }

}

生产者开始生产东西了... 1

生产者开始生产东西了... 2

生产者开始生产东西了... 3

生产者开始生产东西了... 4

消费者开始消费东西了....  4

生产者开始生产东西了... 5

消费者开始消费东西了....  5

生产者开始生产东西了... 6

消费者开始消费东西了....  6

生产者开始生产东西了... 7

消费者开始消费东西了....  7

生产者开始生产东西了... 8

消费者开始消费东西了....  8

生产者开始生产东西了... 9

消费者开始消费东西了....  9

生产者开始生产东西了... 10