使用BlockingQueue构建生产者消费者模式(java并发编程5.3)

拼搏现实的明天。 2022-07-12 04:15 186阅读 0赞

生产者消费者模式

以缓冲区作为生产者和消费者之间沟通的桥梁: 生产者只负责生产, 将生产出来的数据存入缓冲区. 而消费者只负责消费, 不断的从缓冲区中取出数据进行处理.

生产者消费者模式是非常常用的, 因为应用该模式有效的解耦了生产者和消费者. 生产者不需要知道有没有其他生产者在生产, 也不需要知道有多少个消费者在消费, 而消费者不需要知道数据来自哪个生产者. 另外该模式支持并发操作, 如果生产者直接调用消费者的方法, 生产者就必须等到消费者处理完毕才能返回, 万一消费者处理的速度很慢, 就会白白浪费生产者的时间. 而使用模式的话, 生产者只需要将数据存入缓冲区就可以了.

缓冲区是生产者消费者模式的核心. 生产者将数据存入缓冲区的一端, 消费者则负责从缓冲区的另一端取出数据进行处理, 队列非常适用这样的场景. 由于生产者消费者大多处于不同的线程, 队列就必须是线程安全的--java的BlockingQueue可以满足要求.

BlockingQueue

BlockingQueue的put方法用于将数据放入队列, 如果队列已满, put方法所在的线程将阻塞, 直到队列不满. take方法用于从队列中取出数据, 如果队列为空, take方法所在的线程将阻塞, 直到队列不为空.

Java代码   ![收藏代码][icon_star.png]

1.  **public** **void** put(E e) **throws** InterruptedException \{  
2.      **if** (e == **null**)  
3.          **throw** **new** NullPointerException();  
4.      **final** E\[\] items = **this**.items;  
5.      **final** ReentrantLock lock = **this**.lock;  
6.      // 锁定  
7.      lock.lockInterruptibly();  
8.      **try** \{  
9.          **try** \{  
10.             // 如果队列已满, 就阻塞线程  
11.             **while** (count == items.length)  
12.                 notFull.await();  
13.         \} **catch** (InterruptedException ie) \{  
14.             notFull.signal(); // propagate to non-interrupted thread  
15.             **throw** ie;  
16.         \}  
17.         insert(e);  
18.     \} **finally** \{  
19.         lock.unlock();  
20.     \}  
21. \}  
22.   
23. **private** **void** insert(E x) \{  
24.     items\[putIndex\] = x;  
25.     putIndex = inc(putIndex);  
26.     ++count;  
27.     // 插入数据后唤醒在非空条件上阻塞的线程  
28.     notEmpty.signal();  
29. \}  
30.   
31. **public** E take() **throws** InterruptedException \{  
32.     **final** ReentrantLock lock = **this**.lock;  
33.     // 锁定  
34.     lock.lockInterruptibly();  
35.     **try** \{  
36.         **try** \{  
37.             // 如果队列为空, 就阻塞线程  
38.             **while** (count == 0)  
39.                 notEmpty.await();  
40.         \} **catch** (InterruptedException ie) \{  
41.             notEmpty.signal(); // propagate to non-interrupted thread  
42.             **throw** ie;  
43.         \}  
44.         E x = extract();  
45.         **return** x;  
46.     \} **finally** \{  
47.         lock.unlock();  
48.     \}  
49. \}  
50.   
51. **private** E extract() \{  
52.     **final** E\[\] items = **this**.items;  
53.     E x = items\[takeIndex\];  
54.     items\[takeIndex\] = **null**;  
55.     takeIndex = inc(takeIndex);  
56.     --count;  
57.     // 取出数据后唤醒在notFull条件上阻塞的线程  
58.     notFull.signal();  
59.     **return** x;  
60. \}

offer(E e, long timeout, TimeUnit unit)用于将数据放入队列, 如果队列已满, 将最多等待指定的时间, offer返回true时说明数据成功入队, 否则说明没有成功. poll(long timeout, TimeUnit unit)是与offer配对的方法, 用于从队列中取出数据, 如果队列为空, 最多等待指定的时间, poll返回值为null时说明没有取到数据.

Java代码   ![收藏代码][icon_star.png]

1.  **public** **boolean** offer(E e, **long** timeout, TimeUnit unit) **throws** InterruptedException \{  
2.      **if** (e == **null**)  
3.          **throw** **new** NullPointerException();  
4.      // 获得阻塞的最大时间  
5.      **long** nanos = unit.toNanos(timeout);  
6.      **final** ReentrantLock lock = **this**.lock;  
7.      lock.lockInterruptibly();  
8.      **try** \{  
9.          **for** (;;) \{  
10.             // 如果队列没有满, 则插入数据并返回true  
11.             **if** (count != items.length) \{  
12.                 insert(e);  
13.                 **return** **true**;  
14.             \}  
15.             // 如果剩余的等待时间小于等于0说明等待时间已超过最大值, 此时返回false, 表明插入没有成功  
16.             **if** (nanos <= 0)  
17.                 **return** **false**;  
18.             **try** \{  
19.                 // awaitNanos方法用于阻塞队列, 并返回剩余的时间值  
20.                 nanos = notFull.awaitNanos(nanos);  
21.             \} **catch** (InterruptedException ie) \{  
22.                 notFull.signal(); // propagate to non-interrupted thread  
23.                 **throw** ie;  
24.             \}  
25.         \}  
26.     \} **finally** \{  
27.         lock.unlock();  
28.     \}  
29. \}  
30.   
31. **public** E poll(**long** timeout, TimeUnit unit) **throws** InterruptedException \{  
32.     **long** nanos = unit.toNanos(timeout);  
33.     **final** ReentrantLock lock = **this**.lock;  
34.     lock.lockInterruptibly();  
35.     **try** \{  
36.         **for** (;;) \{  
37.             // 如果队列不为空, 就取出数据然后返回  
38.             **if** (count != 0) \{  
39.                 E x = extract();  
40.                 **return** x;  
41.             \}  
42.             // 如果阻塞时间已过最大时间, 就返回null, 说明没有取到数据  
43.             **if** (nanos <= 0)  
44.                 **return** **null**;  
45.             **try** \{  
46.                 // awaitNanos方法用于阻塞队列, 并返回剩余的时间值  
47.                 nanos = notEmpty.awaitNanos(nanos);  
48.             \} **catch** (InterruptedException ie) \{  
49.                 notEmpty.signal(); // propagate to non-interrupted thread  
50.                 **throw** ie;  
51.             \}  
52.         \}  
53.     \} **finally** \{  
54.         lock.unlock();  
55.     \}  
56. \}

BlockingQueue的容量可以是无限的, 也可以是有限的. 无限容量的BlockingQueue永远也不会发生队列已满的事件.

BlockingQueue的常见实现类有ArrayBlockingQueue, LinkedBlockingQueue, 以及PriorityBlockingQueue等. ArrayBlockingQueue底层使用循环数组实现, LinkedBlockingQueue底层使用链表实现. PriorityBlockingQueue则是一个可排序的阻塞队列, 可以按照元素的自然顺序(元素需要实现Comparable接口)或者指定的Comparator排序.

生产者消费者模式的例子

该例子用于模拟对文件进行索引, 生产者FileCrawler类将待索引的文件放入队列, 消费者Indexer则负责从队列中取出文件进行索引标记.

Java代码   ![收藏代码][icon_star.png]

1.  /\*\* 
2.   \* 生产者, 生产待索引的文件 
3.   \*/  
4.  **public** **class** FileCrawler **implements** Runnable \{  
5.      **private** **final** BlockingQueue<File> fileQueue;  
6.      **private** **final** FileFilter fileFilter;  
7.      **private** **final** File root;  
8.    
9.      **public** FileCrawler(BlockingQueue<File> fileQueue, FileFilter fileFilter, File root) \{  
10.         **this**.fileQueue = fileQueue;  
11.         **this**.fileFilter = fileFilter;  
12.         **this**.root = root;  
13.     \}  
14.   
15.     **public** **void** run() \{  
16.         **try** \{  
17.             crawl(root);  
18.         \} **catch** (InterruptedException e) \{  
19.             Thread.currentThread().interrupt();  
20.         \}  
21.     \}  
22.   
23.     **private** **void** crawl(File root) **throws** InterruptedException \{  
24.         File\[\] entries = root.listFiles(fileFilter);  
25.         **if** (entries != **null**) \{  
26.             **for** (File entry : entries)  
27.                 **if** (entry.isDirectory()) \{  
28.                     // 递归调用  
29.                     crawl(entry);  
30.                 \} **else** **if** (!alreadyIndexed(entry)) \{  
31.                     // 向队列中添加文件, 如果队列是BOUND的, 且队列已满, 则put方法将阻塞, 直到队列不满  
32.                     System.out.println(entry + ": 等待进行索引 by " + Thread.currentThread().getName());  
33.                     fileQueue.put(entry);  
34.                 \}  
35.         \}  
36.     \}  
37.   
38.     **private** **boolean** alreadyIndexed(File entry) \{  
39.         **return** **false**;  
40.     \}  
41. \}  
42.   
43. /\*\* 
44.  \* 消费者, 从队列中取出文件进行索引标记 
45.  \*/  
46. **public** **class** Indexer **implements** Runnable \{   
47.     **private** **final** BlockingQueue<File> queue;   
48.   
49.     **public** Indexer(BlockingQueue<File> queue) \{   
50.         **this**.queue = queue;   
51.     \}   
52.   
53.     **public** **void** run() \{   
54.         **try** \{   
55.             **while** (**true**) \{  
56.                 // 从队列中取出file标记索引. 如果队列为空, take方法将阻塞, 直到队列重新不为空.  
57.                 File file = queue.take();  
58.                 indexFile(file);   
59.                 System.out.println(file + ": 已进行索引 by " + Thread.currentThread().getName());  
60.             \}  
61.         \} **catch** (InterruptedException e) \{   
62.             Thread.currentThread().interrupt();   
63.         \}   
64.     \}  
65.   
66.     **private** **void** indexFile(File file) \{  
67.     \}  
68. \}  
69.   
70. /\*\* 
71.  \* 测试生产者消费者模式 
72.  \*/  
73. **public** **class** FileIndexer \{  
74.     **public** **static** **void** startIndexing(File\[\] roots) \{  
75.         // 创建一个BOUNDED队列, 队列中最大的元素为10个  
76.         BlockingQueue<File> queue = **new** LinkedBlockingQueue<File>(10);   
77.         FileFilter filter = **new** FileFilter() \{   
78.             **public** **boolean** accept(File file) \{ **return** **true**; \}   
79.         \};   
80.        
81.         **for** (**int** i = 0; i < roots.length; i++) \{  
82.             File root = roots\[i\];  
83.             // 启动生产者线程  
84.             **new** Thread(**new** FileCrawler(queue, filter, root), "producer " + i).start();   
85.         \}  
86.        
87.         // 启动3个消费者线程  
88.         **for** (**int** i = 0; i < 3; i++) \{  
89.             **new** Thread(**new** Indexer(queue), "consumer " + i).start();  
90.         \}  
91.     \}  
92.       
93.     **public** **static** **void** main(String\[\] args) \{  
94.         File dir = **new** File("E:\\\\TDDOWNLOAD\\\\mina doc");  
95.         startIndexing(dir.listFiles(**new** FileFilter() \{  
96.             @Override  
97.             **public** **boolean** accept(File pathname) \{  
98.                 **if** (pathname.isDirectory()) \{  
99.                     **return** **true**;  
100.                 \}  
101.                 **return** **false**;  
102.             \}  
103.         \}));  
104.     \}  
105. \}

[icon_star.png]: /images/20220711/2f40af8e6f224d4bb2ff2d9f57ceac01.png