且构网

分享程序员开发的那些事...
且构网 - 分享程序员编程开发的那些事

使用阻塞队列实现生产者-消费者模型

更新时间:2022-09-12 23:38:05

生产者-消费者问题

生产者消费者问题也称作有界缓冲区(bounded-buffer)问题,
是操作系统中一个经典的线程同步问题,问题描述如下:
生产者在生产产品提供给若干个消费者去消费,为了使生产者和消费者能并发执行,在两者之间设置一个具有多个缓冲区,生产者将它生产的产品放入缓冲区中,消费者可以从缓冲区中取走产品进行消费,两个进程共享一个公共的固定大小的缓冲区。
显然生产者和消费者之间必须保持同步,即不允许消费者到一个空的缓冲区中取产品,也不允许生产者向一个已经放入产品的缓冲区中再次投放产品。

经典的方法是使用wait和notify方法在生产者和消费者线程中合作,在队列满了或者队列是空的条件下阻塞,
如果使用阻塞队列作为缓冲数据结构,可以不用考虑同步等操作,直接解决问题。

 

使用阻塞队列实现

Java的阻塞队列(BlockingQueue)隐含的提供了这些控制,不需要使用wait和nofity在生产者和消费者之间通信,查看实现:

 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
/**
 * 使用阻塞队列实现生产者-消费者模型
 * 阻塞队列只允许元素以FIFO的方式来访问
 * @author Bingyue
 *
 */
public class ProducerCustomerPattern {
     
    public static void  main(String[] args) {
 
        //生产者和消费者共享的存储区域
        BlockingQueue<Integer> blockQueue=new LinkedBlockingQueue();
         
        /**
         * 此处外部类访问静态方法调用内部类,必须先创建外部类实例。
         * 如果你不需要内部类对象与其外围类对象之间有联系,那你可以将内部类声明为static。这通常称为静态嵌套类(Static Nested Classes)。
         * 想要理解static应用于内部类时的含义,你就必须记住,普通的内部类对象隐含地保存了一个引用,指向创建它的外围类对象。
         * 然而,当内部类是static的时,就不是这样了。嵌套类意味着:  
         * 1. 要创建嵌套类的对象,并不需要其外围类的对象。 
         * 2. 不能从嵌套类的对象中访问非静态的外围类对象。
         */
        ProducerCustomerPattern ps=new ProducerCustomerPattern();
        //注意创建内部类的方式
        Thread pro=new Thread(ps.new Producer(blockQueue));
        Thread cus=new Thread(ps.new Customer(blockQueue));
         
        pro.start();
        cus.start();
    }
     
    class Producer implements Runnable{
         
        private final BlockingQueue<Integer> queue;
         
        public Producer(BlockingQueue<Integer> queue){
            this.queue=queue;
        }
 
        @Override
        public void run() {
            for(int i=0;i<=10;i++){
                try {
                    System.out.println("Produced:"+i);
                    queue.put(i);
                catch (InterruptedException e) {
                    //对于阻塞的插入和移除方法  put和take都抛出 InterruptedException
                    e.printStackTrace();
                }
            }
        }
         
    }
     
    /**
     * 通过实现Runnable接口线程创建
     * (1).定义一个类实现Runnable接口,重写接口中的run()方法。在run()方法中加入具体的任务代码或处理逻辑。
     * (2).创建Runnable接口实现类的对象。
     * (3).创建一个Thread类的对象,需要封装前面Runnable接口实现类的对象。(接口可以实现多继承)
     * (4).调用Thread对象的start()方法,启动线程
     *
     * 通过继承Thread类创建线程
     * (1).首先定义一个类去继承Thread父类,重写父类中的run()方法。在run()方法中加入具体的任务代码或处理逻辑。
     * (2).直接创建一个ThreadDemo2类的对象,也可以利用多态性,变量声明为父类的类型。
     * (3).调用start方法,线程t启动,隐含的调用run()方法。
     */
    class Customer implements Runnable{
 
        private final BlockingQueue<Integer> queue;
         
        public Customer(BlockingQueue<Integer> queue){
            this.queue=queue;
        }
         
        @Override
        public void run() {
            while(true){
                try {
                    System.out.println("Customed:"+queue.take());
                catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
  

  

 

本文转自邴越博客园博客,原文链接:http://www.cnblogs.com/binyue/p/4580089.html,如需转载请自行联系原作者