Java BlockingQueue示例

Dear 丶 2023-02-15 05:08 7阅读 0赞

![format_png][]![format_png 1][]

在Java中，我们可以使用`BlockingQueue`创建生产者和消费者共享的队列。

1.  生产者–生成数据并将其放入队列。
2.  使用者–从队列中删除数据。

> **注意**  
> 阅读本文以了解什么是[生产者和消费者][Link 1] 。

BlockingQueue实现是线程安全的，可以安全地与多个生产者和多个使用者一起使用。

## 1. BlockingQueue ##

一个简单的`BlockingQueue`示例，生产者生成数据并将其放入队列中，与此同时，消费者从同一队列中获取数据。

1.1生产者–一个`Runnable`对象，将20个整数放入队列中。

ExecutorExample1.java

package com.mkyong.concurrency.queue.simple.raw;
    
    import java.util.concurrent.BlockingQueue;
    
    public class Producer implements Runnable {
    
        private final BlockingQueue<Integer> queue;
    
        @Override
        public void run() {
    
            try {
                process();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
    
        }
    
        private void process() throws InterruptedException {
    
            // Put 20 ints into Queue
            for (int i = 0; i < 20; i++) {
                System.out.println("[Producer] Put : " + i);
                queue.put(i);
                System.out.println("[Producer] Queue remainingCapacity : " + queue.remainingCapacity());
                Thread.sleep(100);
            }
    
        }
    
        public Producer(BlockingQueue<Integer> queue) {
            this.queue = queue;
        }
    }

1.2消费品-甲`Runnable`对象采取从队列项目。

Consumer.java

package com.mkyong.concurrency.queue.simple.raw;
    
    import java.util.concurrent.BlockingQueue;
    
    public class Consumer implements Runnable {
    
        private final BlockingQueue<Integer> queue;
    
        @Override
        public void run() {
    
            try {
                while (true) {
                    Integer take = queue.take();
                    process(take);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
    
        }
    
        private void process(Integer take) throws InterruptedException {
            System.out.println("[Consumer] Take : " + take);
            Thread.sleep(500);
        }
    
        public Consumer(BlockingQueue<Integer> queue) {
            this.queue = queue;
        }
    }

1.3运行。 启动1个生产者和1个消费者，并创建一个大小为10的队列。

Main.java

package com.mkyong.concurrency.queue.simple;
    
    import com.mkyong.concurrency.queue.simple.raw.Consumer;
    import com.mkyong.concurrency.queue.simple.raw.Producer;
    
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    
    public class Main {
    
        public static void main(String[] args) {
    
            BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);
    
            new Thread(new Producer(queue)).start();
            new Thread(new Consumer(queue)).start();
    
        }
    
    }

输出量

如果数据已满，生产者将不会尝试将更多数据放入队列。

[Producer] Put : 0
    [Producer] Queue remainingCapacity : 9
    [Consumer] Take : 0
    [Producer] Put : 1
    [Producer] Queue remainingCapacity : 9
    [Producer] Put : 2
    [Producer] Queue remainingCapacity : 8
    [Producer] Put : 3
    [Producer] Queue remainingCapacity : 7
    [Producer] Put : 4
    [Producer] Queue remainingCapacity : 6
    [Producer] Put : 5
    [Producer] Queue remainingCapacity : 5
    [Consumer] Take : 1
    [Producer] Put : 6
    [Producer] Queue remainingCapacity : 5
    [Producer] Put : 7
    [Producer] Queue remainingCapacity : 4
    [Producer] Put : 8
    [Producer] Queue remainingCapacity : 3
    [Producer] Put : 9
    [Producer] Queue remainingCapacity : 2
    [Producer] Put : 10
    [Producer] Queue remainingCapacity : 1
    [Consumer] Take : 2
    [Producer] Put : 11
    [Producer] Queue remainingCapacity : 1
    [Producer] Put : 12
    [Producer] Queue remainingCapacity : 0
    [Producer] Put : 13
    [Consumer] Take : 3
    [Producer] Queue remainingCapacity : 0
    [Producer] Put : 14
    [Consumer] Take : 4
    [Producer] Queue remainingCapacity : 0
    [Producer] Put : 15
    [Consumer] Take : 5
    [Producer] Queue remainingCapacity : 0
    [Producer] Put : 16
    [Consumer] Take : 6
    [Producer] Queue remainingCapacity : 0
    [Producer] Put : 17
    [Consumer] Take : 7
    [Producer] Queue remainingCapacity : 0
    [Producer] Put : 18
    [Consumer] Take : 8
    [Producer] Queue remainingCapacity : 0
    [Producer] Put : 19
    [Consumer] Take : 9
    [Producer] Queue remainingCapacity : 0
    [Consumer] Take : 10
    [Consumer] Take : 11
    [Consumer] Take : 12
    [Consumer] Take : 13
    [Consumer] Take : 14
    [Consumer] Take : 15
    [Consumer] Take : 16
    [Consumer] Take : 17
    [Consumer] Take : 18
    [Consumer] Take : 19

该程序不会停止或退出，它将继续在那里运行，以从`BlockingQueue`放入和取出数据

## 2. BlockingQueue +毒药 ##

“毒丸”是停止或中断生产者线程和使用者线程的一般解决方案。 想法是生产者将“毒丸”放入队列并退出，如果“消费者”看到“毒丸”，则停止并退出。

2.1使用毒丸溶液的生产者。

ProducerPoison.java

package com.mkyong.concurrency.queue.simple.poison;
    
    import java.util.concurrent.BlockingQueue;
    
    public class ProducerPoison implements Runnable {
    
        private final BlockingQueue<Integer> queue;
        private final Integer POISON;
    
        @Override
        public void run() {
    
            try {
                process();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                while (true) {
                    try {
                        queue.put(POISON);
                        break;
                    } catch (InterruptedException e) {
                        //...
                    }
                }
            }
    
        }
    
        private void process() throws InterruptedException {
    
            // Put 20 elements into Queue
            for (int i = 0; i < 20; i++) {
                System.out.println("[Producer] Put : " + i);
                queue.put(i);
                System.out.println("[Producer] Queue remainingCapacity : " + queue.remainingCapacity());
                Thread.sleep(100);
            }
    
        }
    
        public ProducerPoison(BlockingQueue<Integer> queue, Integer POISON) {
            this.queue = queue;
            this.POISON = POISON;
        }
    
    }

2.2有毒药丸溶液的消费者。

ConsumerPoison.java

package com.mkyong.concurrency.queue.simple.poison;
    
    import java.util.concurrent.BlockingQueue;
    
    public class ConsumerPoison implements Runnable {
    
        private final BlockingQueue<Integer> queue;
        private final Integer POISON;
    
        @Override
        public void run() {
    
            try {
                while (true) {
                    Integer take = queue.take();
                    process(take);
    
                    // if this is a poison pill, break, exit
                    if (take == POISON) {
                        break;
                    }
    
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
    
        }
    
        private void process(Integer take) throws InterruptedException {
            System.out.println("[Consumer] Take : " + take);
            Thread.sleep(500);
        }
    
        public ConsumerPoison(BlockingQueue<Integer> queue, Integer POISON) {
            this.queue = queue;
            this.POISON = POISON;
        }
    }

2.3开始2个生产者和2个消费者。

Main.java

package com.mkyong.concurrency.queue.simple;
    
    import com.mkyong.concurrency.queue.simple.poison.ConsumerPoison;
    import com.mkyong.concurrency.queue.simple.poison.ProducerPoison;
    
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    
    public class Main {
    
        public static void main(String[] args) {
    
            BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);
    
            //new Thread(new Producer(queue)).start();
            //new Thread(new Consumer(queue)).start();
    
            Integer poison = -1;
            new Thread(new ProducerPoison(queue, poison)).start();
            new Thread(new ProducerPoison(queue, poison)).start();
    
            new Thread(new ConsumerPoison(queue, poison)).start();
            new Thread(new ConsumerPoison(queue, poison)).start();
    
        }
    
    }

输出量

[Producer] Put : 0
    [Producer] Put : 0
    [//...
    [Consumer] Take : 18
    [Consumer] Take : 18
    [Consumer] Take : 19
    [Consumer] Take : 19
    [Consumer] Take : -1
    [Consumer] Take : -1
    
    Process finished with exit code 0

## 3. BlockingQueue +文件索引 ##

一个`BlockingQueue`示例，用于创建简单的文件索引引擎。 生产者对目录进行爬网，然后将文件名放入队列中，与此同时，消费者从同一队列中获取文件名并为其建立索引。

3.1生产者。

FileCrawlerProducer.java

package com.mkyong.concurrency.queue.crawler;
    
    import java.io.File;
    import java.io.FileFilter;
    import java.util.concurrent.BlockingQueue;
    
    // Producer
    // Crawl file system and put the filename in BlockingQueue.
    public class FileCrawlerProducer implements Runnable {
    
        private final BlockingQueue<File> fileQueue;
        private final FileFilter fileFilter;
        private final File file;
        private final File POISON;
        private final int N_POISON_PILL_PER_PRODUCER;
    
        @Override
        public void run() {
    
            try {
                crawl(file);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                while (true) {
                    try {
                        System.out.println(Thread.currentThread().getName()
                                + " - FileCrawlerProducer is done, try poison all the consumers!");
                        // poison all threads
                        for (int i = 0; i < N_POISON_PILL_PER_PRODUCER; i++) {
                            System.out.println(Thread.currentThread().getName() + " - puts poison pill!");
                            fileQueue.put(POISON);
                        }
                        break;
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
    
        }
    
        public FileCrawlerProducer(BlockingQueue<File> fileQueue,
                                   FileFilter fileFilter, File file, 
                                   File POISON, int n_POISON_PILL_PER_PRODUCER) {
            this.fileQueue = fileQueue;
            this.fileFilter = fileFilter;
            this.file = file;
            this.POISON = POISON;
            N_POISON_PILL_PER_PRODUCER = n_POISON_PILL_PER_PRODUCER;
        }
    
        private void crawl(File root) throws InterruptedException {
    
            File[] entries = root.listFiles(fileFilter);
            if (entries != null) {
                for (File entry : entries) {
                    if (entry.isDirectory()) {
                        crawl(entry);
                    } else if (!isIndexed(entry)) {
                        System.out.println("[FileCrawlerProducer] - Found..." 
                                + entry.getAbsoluteFile());
                        fileQueue.put(entry);
                    }
                }
            }
    
        }
    
        private boolean isIndexed(File f) {
            return false;
        }
    
    }

3.2消费者。

IndexerConsumer.java

package com.mkyong.concurrency.queue.crawler;
    
    import java.io.File;
    import java.util.concurrent.BlockingQueue;
    
    // Consumer
    public class IndexerConsumer implements Runnable {
    
        private final BlockingQueue<File> fileQueue;
        private final File POISON;
    
        @Override
        public void run() {
    
            try {
                while (true) {
                    File take = fileQueue.take();
                    if (take == POISON) {
                        System.out.println(Thread.currentThread().getName() + " die");
                        break;
                    }
                    indexFile(take);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
    
        }
    
        public void indexFile(File file) {
            if (file.isFile()) {
                System.out.println(Thread.currentThread().getName() 
                        + " [IndexerConsumer] - Indexing..." + file.getAbsoluteFile());
            }
    
        }
    
        public IndexerConsumer(BlockingQueue<File> fileQueue, File POISON) {
            this.fileQueue = fileQueue;
            this.POISON = POISON;
        }
    }

3.3开始1个生产者和2个消费者。

Main.java

package com.mkyong.concurrency.queue.crawler;
    
    import java.io.File;
    import java.io.FileFilter;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    
    public class Main {
    
        private static final File POISON = new File("This is a POISON PILL");
    
        public static void main(String[] args) {
    
            int N_PRODUCERS = 1;
            int N_CONSUMERS = 2;//Runtime.getRuntime().availableProcessors();
            int N_POISON_PILL_PER_PRODUCER = N_CONSUMERS / N_PRODUCERS;
            int N_POISON_PILL_REMAIN = N_CONSUMERS % N_PRODUCERS;
    
            System.out.println("N_PRODUCERS : " + N_PRODUCERS);
            System.out.println("N_CONSUMERS : " + N_CONSUMERS);
            System.out.println("N_POISON_PILL_PER_PRODUCER : " + N_POISON_PILL_PER_PRODUCER);
            System.out.println("N_POISON_PILL_REMAIN : " + N_POISON_PILL_REMAIN);
    
            //unbound queue, no limit
            BlockingQueue<File> queue = new LinkedBlockingQueue<>();
    
            FileFilter filter = new FileFilter() {
                public boolean accept(File file) {
                    return true;
                }
            };
    
            File root = new File("C:\\users");
    
            for (int i = 0; i < N_PRODUCERS - 1; i++) {
                new Thread(new FileCrawlerProducer(queue, filter, root,
                        POISON, N_POISON_PILL_PER_PRODUCER)).start();
            }
            new Thread(new FileCrawlerProducer(queue, filter, root, POISON,
                    N_POISON_PILL_PER_PRODUCER + N_POISON_PILL_REMAIN)).start();
    
            for (int i = 0; i < N_CONSUMERS; i++) {
                new Thread(new IndexerConsumer(queue, POISON)).start();
            }
    
        }
    }

输出量

//...
    [FileCrawlerProducer] - Found...C:\users\Public\Videos\desktop.ini
    Thread-2 [IndexerConsumer] - Indexing...C:\users\Public\Videos\desktop.ini
    Thread-0 - FileCrawlerProducer is done, try poison all the consumers!
    Thread-0 - puts poison pill!
    Thread-0 - puts poison pill!
    Thread-1 die
    Thread-2 die
    
    Process finished with exit code 0

## 下载源代码 ##

> git clone [https://github.com/mkyong/java-concurrency.git][https_github.com_mkyong_java-concurrency.git]

## 参考文献 ##

1.  [生产者－消费者问题][Link 1]
2.  [BlockingQueue JavaDoc][]

标签： [BlockingQueue][] [并发][Link 2] [Java][] [队列][Link 3] [线程][Link 4]

> 翻译自: [https://mkyong.com/java/java-blockingqueue-examples/][https_mkyong.com_java_java-blockingqueue-examples]

[format_png]: http://imgconvert.csdnimg.cn/aHR0cDovL3d3dy5ta3lvbmcuY29tL3dwLWNvbnRlbnQvdXBsb2Fkcy8yMDE4LzEyL3Byb2R1Y2VyLWNvbnN1bWVyLWJsb2NraW5ncXVldWUtamF2YS5wbmc?x-oss-process=image/format,png
[format_png 1]: http://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3cubWt5b25nLmNvbS93cC1jb250ZW50L3VwbG9hZHMvMjAxOC8xMi9wcm9kdWNlci1jb25zdW1lci1ibG9ja2luZ3F1ZXVlLWphdmEucG5n?x-oss-process=image/format,png
[Link 1]: https://en.wikipedia.org/wiki/Producer%E2%80%93consumer_problem
[https_github.com_mkyong_java-concurrency.git]: https://github.com/mkyong/java-concurrency.git
[BlockingQueue JavaDoc]: https://docs.oracle.com/javase/8/docs/api/?java/util/concurrent/BlockingQueue.html
[BlockingQueue]: https://mkyong.com/tag/blockingqueue/
[Link 2]: https://mkyong.com/tag/concurrency/
[Java]: https://mkyong.com/tag/java/
[Link 3]: https://mkyong.com/tag/queue/
[Link 4]: https://mkyong.com/tag/thread/
[https_mkyong.com_java_java-blockingqueue-examples]: https://mkyong.com/java/java-blockingqueue-examples/