BlockingQueue生产者消费者代码实现

题目：编写一段生产者/消费者的 Java 代码，其中生产者每次生产 1 个 0 到 1000 之间的随机数，消费者则把该随机数打印出来。如果产生的随机数为 0，则生产者、消费者均退出运行。要求生产者、消费者均使用线程来实现。

//方法一：
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ThreadTest {
    public static void main(String[] args) {
        Storage storage = new Storage();
        new Producer(storage).start();
        new Customer(storage).start();

    }

}

class Customer extends Thread {

    private Storage storage;

    public Customer(Storage storage) {
        this.storage = storage;
    }

    public void run() {
        while (true) {
            try {
                if (Storage.NUM == 0) {
                    System.out.println("打印：" + 0 + ",线程终止");
                    break;
                }
                Integer num = storage.print();
                System.out.println("打印：" + num);
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

class Producer extends Thread {

    private Storage storage;

    public Producer(Storage storage) {
        this.storage = storage;
    }

    public void run() {

        while (true) {

            try {
                Integer num = new Random().nextInt(1000);
                Storage.NUM = num;
                if (Storage.NUM == 0) {
                    System.out.println("生产：" + 0 + ",线程终止");
                    break;
                }
                storage.put(num);
                System.out.println("生产：" + num);
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }

    }
}

class Storage {
    private BlockingQueue<Integer> randomData = new ArrayBlockingQueue<Integer>(
            1);// 空间大小为1的阻塞队列

    public static int NUM = -1;

    public void put(Integer number) {

        try {
            randomData.put(number);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    public int print() {
        try {
            Integer number = randomData.take();
            return number;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 0;
    }
}


//方法二：
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;

public class Test {

    BlockingQueue<Integer> queue = new SynchronousQueue<Integer>();

    Random rd = new Random();

    class Producer extends Thread {
        @Override
        public void run() {
            while (true) {
                int number = rd.nextInt(1000);
                System.out.println("Producer Generate : " + number);
                try {
                    queue.put(number);
                    if (number == 0) {
                        System.out.println("Producer stoped.");
                        return;
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

        }

    }

    class Consumer extends Thread {
        @Override
        public void run() {
            while (true) {
                try {
                    if (queue.take() == 0) {
                        System.out.println("Consumer stoped.");
                        return;
                    }
                    ;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public void startTest() {
        Producer producer = new Producer();
        Consumer consumer = new Consumer();
        producer.start();
        consumer.start();
    }

    public static void main(String[] args) {
        new Test().startTest();
    }
}


//方法三：
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class BlockingQueueTest {
    private static String FLAG="99";

    public static class Basket {
        // 篮子，能够容纳1000个数字
        BlockingQueue<String> basket = new LinkedBlockingQueue<String>(1000);

        // 生产数字，放入篮子
        public String produce() throws InterruptedException {
            int num = new Random().nextInt(1000);
            System.out.println("++放入篮子中一个数字："+num );
            basket.put(num+"");
            return num+"";
        }

        // 消费数字，从篮子中取走
        public String consume() throws InterruptedException {
            String num=basket.take();
            System.out.println("--取出篮子中一个数字："+num );
            return num;
        }

    }


    // 　测试方法
    public static void testBasket() {
        final Basket basket = new Basket();
        // 定义生产者
        class Producer implements Runnable {
            public void run() {
                try {
                    while (true) {
                        if(basket.produce().equals(FLAG)){
                            System.out.println("！成功生产正确数字:"+FLAG);
                            break;
                        }
                        // 休眠300ms
                        Thread.sleep(300);
                    }
                } catch (InterruptedException ex) {
                }
            }
        }

        // 定义数字消费者
        class Consumer implements Runnable {
            public void run() {
                try {
                    while (true) {
                        // 消费数字
                        // 休眠1000ms
                        if(basket.consume().equals(FLAG)){
                            System.out.println("！成功获取正确数字:"+FLAG);
                            break;
                        }
                        Thread.sleep(500);
                    }
                } catch (InterruptedException ex) {
                }
            }
        }

        ExecutorService service = Executors.newCachedThreadPool();
        Producer producer = new Producer();
        Consumer consumer = new Consumer();
        service.submit(producer);
        service.submit(consumer);

        // 程序运行60s后，所有任务停止
        try {
            Thread.sleep(60*1000);
        } catch (InterruptedException e) {
        }


        service.shutdownNow();
    }

    public static void main(String[] args) {
        BlockingQueueTest.testBasket();
    }
}
}