基于Java阻塞队列的搜索实例
基于Java阻塞队列的搜索实例
2012-06-14 10:34 frogong frogong的博客 我要评论(0) 字号：T | T
队列以一种先进先出的方式管理数据，如果你试图向一个已经满了的阻塞队列中添加一个元素，或是从一个空的阻塞队列中移除一个元素，将导致线程阻塞。在多线程进行合作时，阻塞队列是很有用的工具，工作者线程可以定期的把中间结果存到阻塞队列中，而其他工作者线程把中间结果取出并在将来修改它们。
AD：
队列以一种先进先出的方式管理数据。如果你试图向一个已经满了的阻塞队列中添加一个元素，或是从一个空的阻塞队列中移除一个元素，将导致线程阻塞。在多线程进行合作时，阻塞队列是很有用的工具。工作者线程可以定期的把中间结果存到阻塞队列中。而其他工作者线程把中间结果取出并在将来修改它们。队列会自动平衡负载。如果第一个线程集运行的比第二个慢，则第二个线程集在等待结果时就会阻塞。如果第一个线程集运行的快，那么它将等待第二个线程集赶上来。
下面的程序展示了如何使用阻塞队列来控制线程集。程序在一个目录及它的所有子目录下搜索所有文件，打印出包含指定关键字的文件列表。
java.util.concurrent包提供了阻塞队列的4个变种：LinkedBlockingQueue、ArrayBlockingQueue、PriorityBlockingQueue和DelayQueue。我们用的是ArrayBlockingQueue。ArrayBlockingQueue在构造时需要给定容量，并可以选择是否需要公平性。如果公平参数被设置了，等待时间最长的线程会优先得到处理。通常，公平性会使你在性能上付出代价，只有在的确非常需要的时候再使用它。
生产者线程枚举在所有子目录下的所有文件并把它们放到一个阻塞队列中。这个操作很快，如果队列没有设上限的话，很快它就包含了没有找到的文件。
我们同时还启动了大量的搜索线程。每个搜索线程从队列中取出一个文件，打开它，打印出包含关键字的所有行，然后取出下一个文件。我们使用了一个小技巧来在工作结束后终止线程。为了发出完成信号，枚举线程把一个虚拟对象放入队列。（这类似于在行李输送带上放一个写着“最后一个包”的虚拟包。）当搜索线程取到这个虚拟对象时，就将其放回并终止。
注意，这里不需要人任何显示的线程同步。在这个程序中，我们使用队列数据结构作为一种同步机制。
import java.io.*;
import java.util.*;
import java.util.concurrent.*;
public class BlockingQueueTest
{
public static void main(String[] args)
{
Scanner in = new Scanner(System.in);
System.out.print("Enter base directory (e.g. /usr/local/jdk1.6.0/src): ");
String directory = in.nextLine();
System.out.print("Enter keyword (e.g. volatile): ");
String keyword = in.nextLine();
final int FILE_QUEUE_SIZE = 10;
final int SEARCH_THREADS = 100;
BlockingQueue<File> queue = new ArrayBlockingQueue<File>(FILE_QUEUE_SIZE);
FileEnumerationTask enumerator = new FileEnumerationTask(queue, new File(directory));
new Thread(enumerator).start();
for (int i = 1; i <= SEARCH_THREADS; i++)
new Thread(new SearchTask(queue, keyword)).start();
}
}
/**
* This task enumerates all files in a directory and its subdirectories.
*/
class FileEnumerationTask implements Runnable
{
/**
* Constructs a FileEnumerationTask.
* @param queue the blocking queue to which the enumerated files are added
* @param startingDirectory the directory in which to start the enumeration
*/
public FileEnumerationTask(BlockingQueue<File> queue, File startingDirectory)
{
this.queue = queue;
this.startingDirectory = startingDirectory;
}
public void run()
{
try
{
enumerate(startingDirectory);
queue.put(DUMMY);
}
catch (InterruptedException e)
{
}
}
/**
* Recursively enumerates all files in a given directory and its subdirectories
* @param directory the directory in which to start
*/
public void enumerate(File directory) throws InterruptedException
{
File[] files = directory.listFiles();
for (File file : files)
{
if (file.isDirectory()) enumerate(file);
else queue.put(file);
}
}
public static File DUMMY = new File("");
private BlockingQueue<File> queue;
private File startingDirectory;
}
/**
* This task searches files for a given keyword.
*/
class SearchTask implements Runnable
{
/**
* Constructs a SearchTask.
* @param queue the queue from which to take files
* @param keyword the keyword to look for
*/
public SearchTask(BlockingQueue<File> queue, String keyword)
{
this.queue = queue;
this.keyword = keyword;
}
public void run()
{
try
{
boolean done = false;
while (!done)
{
File file = queue.take();
if (file == FileEnumerationTask.DUMMY)
{
queue.put(file);
done = true;
}
else search(file);
}
}
catch (IOException e)
{
e.printStackTrace();
}
catch (InterruptedException e)
{
}
}
/**
* Searches a file for a given keyword and prints all matching lines.
* @param file the file to search
*/
public void search(File file) throws IOException
{
Scanner in = new Scanner(new FileInputStream(file));
int lineNumber = 0;
while (in.hasNextLine())
{
lineNumber++;
String line = in.nextLine().trim();
if (line.contains(keyword)) System.out.printf("%s:%d %s%n", file.getPath(), lineNumber, line);
}
in.close();
}
private BlockingQueue<File> queue;
private String keyword;
}
这是第12周java总结博客的第二篇了 ，主题是阻塞队列
阻塞队列是用来做什么的？
当试图向队列中添加元素而队列已满，或是从队列中移出元素而队列是空的时候，阻塞队列导致线程阻塞，在协调多个线程之间的合作是，阻塞队列是一个有用的工具。
下面关于BlockingQueue的一个例子，不错哟
目的：学会使用阻塞队列来控制线程集，这个程序主要要实现的功能是在一个目录及它的所有的子目录下搜索所有文件，打印出包含指定关键字的行
总结：1.在将该目录及其子目录下的文件（请注意导入的是文件，而不是文件夹或者目录）导入到阻塞队列中要注意的问题
a.将文件插入到阻塞队列中用put函数，这个可以查api
b.search 方法中，短短的几行代码，其实玄机多着呢！非常不明显的递归，只有该文件不是目录时才将文件插入队列，如下
[java] view plaincopy
public void search (File directory) throws InterruptedException {
File[] files = directory.listFiles();
for(File file:files) {
if(file.isDirectory()) search(file);
else queue.put(file);
}
}
2.查找文件中包含关键字的行时，用到了输入流，再将输入流做形参构造Scanner对象，这样子就觉得好方便。
3.查询结束的条件是 当文件是DUMMY（空文件），这是在进行导入是做下的，一个小的技巧哟！蛮不错的
代码一(实现了Runnable接口的一个类，用来启动线程的，这个线程主要是将一个目录及其子目录下的文件导入到阻塞队列中)
[java] view plaincopy
package searchLine;
import java.io.File;
import java.util.concurrent.BlockingQueue;
/**
* 目的：给你一个目录，将该目录及其子目录下的所有的文件都导入到阻塞队列中去
* 关键：递归，线程
* @author sornor
* @version 1.0
*/
public class FileEnumTask implements Runnable {
private BlockingQueue<File> queue; //阻塞队列对象
private File startingDirectory; //根目录
public final static File DUMMY = new File(""); //DUMMY变量作为查找完成的标志了
/**
* 构造函数 ，基本上就是初始化域
* @param queue
* @param start
*/
public FileEnumTask (BlockingQueue<File> queue,File start) {
//DUMMY = new File(""); //空的文件，这个变量的作用要看以后的输出功能了
this.queue = queue;
this.startingDirectory = start;
}
/**
* 重写Runnable的run方法，查找这个目录
*/
public void run() {
try {
search(startingDirectory);
queue.put(DUMMY);
}
catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 查找该目录下的文件，并加入到队列中去 递归的思想体现的淋漓尽至了
* @param directory 查找的目录
* @throws InterruptedException 中断异常
*/
public void search (File directory) throws InterruptedException {
File[] files = directory.listFiles();
for(File file:files) {
if(file.isDirectory()) search(file);
else queue.put(file);
}
}
}
代码二 （又是一个实现了Runnable接口的类，用来查找每一个文件，并将每个文件中包含关键字的行输出来）
[java] view plaincopy
package searchLine;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.BlockingQueue;
/**
* 目的：这个程序用来查找文件内部的关键字，并且将关键字所在行数打印出来
* @author sornor
* @version 1.0
*/
public class SearchLineTask implements Runnable{
private BlockingQueue<File> queue; //阻塞队列
private String keyword; //待匹配的的关键字
/**
* 构造函数，初始化变量
* @param queue
* @param keyword
*/
public SearchLineTask (BlockingQueue<File> queue,String keyword) {
this.queue = queue;
this.keyword = keyword;
}
/**
* 重写Runnable接口的run方法
*/
public void run() {
boolean isend = false;
try {
while(!isend) {
File file = queue.take();
if (file == FileEnumTask.DUMMY) {
//如果到了文件队列的末尾，也就算说队列中只有这个空文件了
queue.put(file);
isend = true;
}
else searchLine(file);
}
}
catch(InterruptedException e) {
e.printStackTrace();
}
catch (IOException e) {
e.printStackTrace();
}
}
/**
* 查找文件中有关键字的行了
* @param file 待查找的文件
* @throws FileNotFoundException
*/
public void searchLine(File file) throws FileNotFoundException {
//下面是一个小技巧，真不错哟
Scanner scanner = new Scanner(new FileInputStream(file));
int rowNumber = 0;
String line = scanner.nextLine();
while(scanner.hasNext()) {
rowNumber++;
if(line.contains(keyword))
System.out.println("在文件"+file.getPath()+"中，"+rowNumber+" 行，如下 "+line);
line = scanner.nextLine();
}
scanner.close();
}
}
代码三（主函数：键盘输入目录和关键字，执行线程）
[java] view plaincopy
package searchLine;
import java.io.File;
import java.util.Scanner;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
* 目的：测试阻塞队列
* 总结：利用线程，递归的思想达到目的
* @author sornor
* @version 1.0
*
*/
public class BlockingQueueTest {
public static void main(String[] args) {
//先从键盘输入文件的目录和关键字
Scanner scanner = new Scanner(System.in);
System.out.println("directory：");
String directory = scanner.nextLine();
System.out.println("keyword: ");
String keyword = scanner.nextLine();
final int FILE_QUEUE_SIZE = 10;
final int SEARCH_THREADS = 100;
BlockingQueue<File> queue = new ArrayBlockingQueue<File>(FILE_QUEUE_SIZE);
FileEnumTask fet = new FileEnumTask(queue,new File(directory));
new Thread(fet).start();//执行run函数，将所有文件导入到阻塞队列中
for(int i = 1;i < SEARCH_THREADS;i++) { //阻塞队列协调多个线程之间的合作
new Thread(new SearchLineTask(queue,keyword)).start();
}
}
}
http://www.iteye.com/topic/483115
package com.test;
import java.awt.BorderLayout;
import java.awt.Color;
import java.awt.Component;
import java.awt.Dimension;
import java.awt.EventQueue;
import java.awt.Font;
import java.awt.GradientPaint;
import java.awt.Graphics;
import java.awt.Graphics2D;
import java.awt.GridLayout;
import java.awt.Insets;
import java.awt.Rectangle;
import java.awt.event.ComponentAdapter;
import java.awt.event.ComponentEvent;
import java.awt.event.MouseAdapter;
import java.awt.event.MouseEvent;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import javax.swing.BorderFactory;
import javax.swing.ButtonGroup;
import javax.swing.JCheckBox;
import javax.swing.JComponent;
import javax.swing.JFrame;
import javax.swing.JLabel;
import javax.swing.JPanel;
import javax.swing.JRadioButton;
import javax.swing.JScrollPane;
import javax.swing.JSplitPane;
import javax.swing.SpringLayout;
import javax.swing.WindowConstants;
import javax.swing.border.Border;
public class ExampleAccordion extends JPanel {
private static final long serialVersionUID = 1L;
private final JPanel panel = new JPanel();
private final JLabel label = new JLabel();
//分割窗体
private final JSplitPane split = new JSplitPane();
private final JScrollPane scroll;
//折叠效果
public ExampleAccordion() {
super(new BorderLayout());
panel.setOpaque(true);
panel.setBackground(new Color(116, 149, 226));
//滚动条
scroll = new JScrollPane(JScrollPane.VERTICAL_SCROLLBAR_AS_NEEDED,
JScrollPane.HORIZONTAL_SCROLLBAR_NEVER);
scroll.getVerticalScrollBar().setUnitIncrement(10);
scroll.getViewport().add(panel);
//构建数据列表
List panelList = makeList();
//设定监听
accordionListener exr = new accordionListener() {
public void accordionStateChanged(accordionEvent e) {
initComponent();
}
};
for (Iterator it = panelList.iterator(); it.hasNext();) {
AccordionPanel epl = (AccordionPanel) it.next();
addComponent(epl);
epl.addaccordionListener(exr);
}
//加载滚动条监听
scroll.getViewport().addComponentListener(new ComponentAdapter() {
public void componentResized(ComponentEvent e) {
initComponent();
}
});
// 设定大小
label.setPreferredSize(new Dimension(200, 260));
scroll.setPreferredSize(new Dimension(200, 260));
scroll.setMinimumSize(new Dimension(200, 260));
split.setLeftComponent(scroll);
split.setRightComponent(label);
split.setDividerSize(1);
split.setBackground(Color.WHITE);
add(split, BorderLayout.CENTER);
}
public void initComponent() {
Rectangle re = scroll.getViewport().getViewRect();
Insets ins = panel.getInsets();
int cw = (int) re.getWidth() - ins.left - ins.right - 20;
int ch = 10;
Component[] list = panel.getComponents();
for (int i = 0; i < list.length; i++) {
JComponent tmp = (JComponent) list[i];
int th = tmp.getPreferredSize().height;
tmp.setPreferredSize(new Dimension(cw, th));
ch = ch + th + 10;
}
panel.setPreferredSize(new Dimension((int) re.getWidth(), ch + ins.top
+ ins.bottom));
panel.revalidate();
}
public void addComponent(Component label) {
SpringLayout layout = new SpringLayout();
Component[] list = panel.getComponents();
if (list.length == 0) {
layout.putConstraint(SpringLayout.WEST, label, 10,
SpringLayout.WEST, panel);
layout.putConstraint(SpringLayout.NORTH, label, 10,
SpringLayout.NORTH, panel);
} else {
JComponent cmp = null;
for (int i = 0; i < list.length; i++) {
JComponent tmp = (JComponent) list[i];
layout.putConstraint(SpringLayout.WEST, tmp, 10,
SpringLayout.WEST, panel);
if (cmp == null) {
layout.putConstraint(SpringLayout.NORTH, tmp, 10,
SpringLayout.NORTH, panel);
} else {
layout.putConstraint(SpringLayout.NORTH, tmp, 10,
SpringLayout.SOUTH, cmp);
}
cmp = tmp;
}
layout.putConstraint(SpringLayout.WEST, label, 10,
SpringLayout.WEST, panel);
layout.putConstraint(SpringLayout.NORTH, label, 10,
SpringLayout.SOUTH, cmp);
}
panel.add(label);
panel.setLayout(layout);
initComponent();
}
private List<AccordionPanel> makeList() {
List<AccordionPanel> panelList = new ArrayList<AccordionPanel>();
panelList.add(new AccordionPanel("列表1") {
private static final long serialVersionUID = 1L;
public JPanel makePanel() {
JPanel pnl = new JPanel(new GridLayout(0, 1)); //在这个panel中用网格布局，且只显示一列，所有列相同宽度。
JCheckBox c1 = new JCheckBox("aaaaaa");
JCheckBox c2 = new JCheckBox("bbbbbb");
c1.setOpaque(false);
c2.setOpaque(false);
pnl.add(c1);
pnl.add(c2);
pnl.setSize(new Dimension(0, 60));
pnl.setBorder(BorderFactory.createEmptyBorder(5, 15, 5, 15));
return pnl;
}
});
panelList.add(new AccordionPanel("列表2") {
private static final long serialVersionUID = 1L;
public JPanel makePanel() {
JPanel pnl = new JPanel(new GridLayout(0, 1));
pnl.add(new JLabel("辛苦遭逢起一经"));
pnl.add(new JLabel("干戈寥落四周星"));
pnl.add(new JLabel("山河破碎风飘絮"));
pnl.add(new JLabel("身世浮沉雨打萍"));
pnl.setSize(new Dimension(0, 100));
pnl.setBorder(BorderFactory.createEmptyBorder(5, 15, 5, 15));
return pnl;
}
});
panelList.add(new AccordionPanel("列表3") {
private static final long serialVersionUID = 1L;
public JPanel makePanel() {
JPanel pnl = new JPanel(new GridLayout(0, 1));
JRadioButton b1 = new JRadioButton("aa");
JRadioButton b2 = new JRadioButton("bb");
JRadioButton b3 = new JRadioButton("cc");
b1.setOpaque(false);
b2.setOpaque(false);
b3.setOpaque(false);
pnl.add(b1);
pnl.add(b2);
pnl.add(b3);
ButtonGroup bg = new ButtonGroup();
bg.add(b1);
bg.add(b2);
bg.add(b3);
b1.setSelected(true);
pnl.setSize(new Dimension(0, 80));
pnl.setBorder(BorderFactory.createEmptyBorder(5, 15, 5, 15));
return pnl;
}
});
panelList.add(new AccordionPanel("列表4") {
private static final long serialVersionUID = 1L;
public JPanel makePanel() {
JPanel pnl = new JPanel(new GridLayout(0, 1));
JRadioButton b1 = new JRadioButton("aa");
JRadioButton b2 = new JRadioButton("bb");
JRadioButton b3 = new JRadioButton("cc");
b1.setOpaque(false);
b2.setOpaque(false);
b3.setOpaque(false);
pnl.add(b1);
pnl.add(b2);
pnl.add(b3);
ButtonGroup bg = new ButtonGroup();
bg.add(b1);
bg.add(b2);
bg.add(b3);
b1.setSelected(true);
pnl.setSize(new Dimension(0, 80));
pnl.setBorder(BorderFactory.createEmptyBorder(5, 15, 5, 15));
return pnl;
}
});
return panelList;
}
public static void main(String[] args) {
EventQueue.invokeLater(new Runnable() {
public void run() {
createUI();
}
});
}
public static void createUI() {
JFrame frame = new JFrame("JAVA实现类Windows导航栏");
frame.setDefaultCloseOperation(WindowConstants.EXIT_ON_CLOSE);
frame.getContentPane().add(new ExampleAccordion());
frame.pack();
frame.setLocationRelativeTo(null);
frame.setVisible(true);
}
}
class accordionEvent extends java.util.EventObject {
private static final long serialVersionUID = 1L;
public accordionEvent(Object source) {
super(source);
}
}
interface accordionListener {
public void accordionStateChanged(accordionEvent e);
}
abstract class AccordionPanel extends JPanel {
private static final long serialVersionUID = 1L;
abstract public JPanel makePanel();
private final String _title;
private final JLabel label;
private final JPanel panel;
private boolean openFlag = false;
public AccordionPanel(String title) {
super(new BorderLayout());
_title = title;
label = new JLabel("↓ " + title) {
private static final long serialVersionUID = 1L;
protected void paintComponent(Graphics g) {
Graphics2D g2 = (Graphics2D) g;
//绘制渐变
g2.setPaint(new GradientPaint(50, 0, Color.WHITE, getWidth(),
getHeight(), new Color(199, 212, 247)));
g2.fillRect(0, 0, getWidth(), getHeight());
super.paintComponent(g);
}
};
label.addMouseListener(new MouseAdapter() {
public void mousePressed(MouseEvent evt) {
openFlag = !openFlag;
initPanel();
fireaccordionEvent();
}
});
label.setForeground(new Color(33, 93, 198));
label.setFont(new Font("宋体", 1, 12));
label.setBorder(BorderFactory.createEmptyBorder(2, 5, 2, 2));
panel = makePanel();
panel.setOpaque(true);
Border outBorder = BorderFactory.createMatteBorder(0, 2, 2, 2,Color.WHITE);
Border inBorder = BorderFactory.createEmptyBorder(10, 10, 10, 10);
Border border = BorderFactory.createCompoundBorder(outBorder, inBorder);
panel.setBorder(border);
panel.setBackground(new Color(240, 240, 255));
add(label, BorderLayout.NORTH);
}
public boolean isSelected() {
return openFlag;
}
public void setSelected(boolean flg) {
openFlag = flg;
initPanel();
}
protected void initPanel() {
if (isSelected()) {
label.setText("↑ " + _title);
add(panel, BorderLayout.CENTER);
setPreferredSize(new Dimension(getSize().width,
label.getSize().height + panel.getSize().height));
revalidateParent();
} else {
label.setText("↓ " + _title);
remove(panel);
setPreferredSize(new Dimension(getSize().width,
label.getSize().height));
}
revalidate(); //刷新界面
}
private JFrame getParentDialog() {
Component c = this.getParent();
if(c instanceof JFrame)
{
return (JFrame)c;
}
else
return getParentDialog(c);
}
protected ArrayList<accordionListener> accordionListenerList = new ArrayList<accordionListener>();
public void addaccordionListener(accordionListener listener) {
if (!accordionListenerList.contains(listener))
accordionListenerList.add(listener);
}
public void removeaccordionListener(accordionListener listener) {
accordionListenerList.remove(listener);
}
@SuppressWarnings("unchecked")
public void fireaccordionEvent() {
List<accordionListener> list = (List<accordionListener>) accordionListenerList.clone();
Iterator<accordionListener> it = list.iterator();
accordionEvent e = new accordionEvent(this);
while (it.hasNext()) {
accordionListener listener = (accordionListener) it.next();
listener.accordionStateChanged(e);
}
}
}
