我构建了一个基本的web解析器，它使用hadoop将url传递给多个线程。在我到达输入文件的末尾之前，hadoop会在仍有线程运行的情况下声明自己已经完成了。这将导致错误org.apache.hadoop.fs.fserror:java.io.ioexception:stream closed。不管怎样，是否有足够长的时间来保持流的开放以便线程结束(我可以合理准确地预测线程在单个url上花费的最大时间）。
下面是我如何执行线程

public static class Map extends MapReduceBase implements
            Mapper<LongWritable, Text, Text, Text> {
        private Text word = new Text();
        private URLPile pile = new URLPile();
        private MSLiteThread[] Threads = new MSLiteThread[16];
        private boolean once = true;

        @Override
        public void map(LongWritable key, Text value,
                OutputCollector<Text, Text> output, Reporter reporter) {

            String url = value.toString();
            StringTokenizer urls = new StringTokenizer(url);
            Config.LoggerProvider = LoggerProvider.DISABLED;
             System.out.println("In Mapper");
            if (once) {
                for (MSLiteThread thread : Threads) {
                    System.out.println("created thread");
                    thread = new MSLiteThread(pile);
                    thread.start();
                }
                once = false;
            }

            while (urls.hasMoreTokens()) {
                try {
                    word.set(urls.nextToken());
                    String currenturl = word.toString();
                    pile.addUrl(currenturl, output);

                } catch (Exception e) {
                    e.printStackTrace();
                    continue;
                }

            }

        }

线程本身得到如下url

public void run(){
            try {
            sleep(3000);
                while(!done()){
                    try {
                    System.out.println("in thread");
                      MSLiteURL tempURL = pile.getNextURL();
                      String currenturl = tempURL.getURL();
                      urlParser.parse(currenturl);
                      urlText.set("");
                      titleText.set(currenturl+urlParser.export());
                      System.out.println(urlText.toString()+titleText.toString());
                      tempURL.getOutput().collect(urlText, titleText);
                      pile.doneParsing();
                     sleep(30);
                    } catch (Exception e) {
                          pile.doneParsing();
                    e.printStackTrace();
                        continue;
                    }
                }
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            System.out.println("Thread done");

        }

并介绍了urlpile中的相关方法

public synchronized void addUrl(String url,OutputCollector<Text, Text> output) throws InterruptedException {
        while(queue.size()>16){
            System.out.println("queue full");
            wait();
        }
        finishedParcing--;
        queue.add(new MSLiteURL(output,url));
        notifyAll();
    }

    private Queue<MSLiteURL> queue = new LinkedList<MSLiteURL>();
    private int sent = 0;
    private int finishedParcing = 0;
    public synchronized MSLiteURL getNextURL() throws InterruptedException {

        notifyAll();
        sent++;
        //System.out.println(queue.peek());
        return queue.remove();

    }