分步骤批量处理数据，每个Pipe处理完所有数据再进行下一个Pipe

@param reader
@throws Exception
551382
public void loadThruStagePipes(Reader reader) throws Exception {
        SeriesPipes p = (SeriesPipes) pipes;
        // 通过迭代加入样本
        Pipe p1 = p.getPipe(0);
        while (reader.hasNext()) {
            Instance inst = reader.next();
            if(inst!=null){
                if (p1 != null)
                    p1.addThruPipe(inst);
                this.add(inst);
               
            };
        }
        for (int i = 1; i < p.size(); i++)
            p.getPipe(i).process(this);
    }
326508
public Reader readRaw() throws IOException {
        try {
            InputStream fileInputStream = Files.newInputStream(file.toPath());
            try {
                return new InputStreamReader(fileInputStream, sniffEncoding());
            } catch (IOException ex) {
                // Exception may happen if we fail to find encoding or if this encoding is unsupported.
                // In such case we close the underlying stream and rethrow.
                Util.closeAndLogFailures(fileInputStream, LOGGER, "FileInputStream", file.toString());
                throw ex;
            }
        } catch (InvalidPathException e) {
            throw new IOException(e);
        }
    }
1125540
public void process(File file) throws Exception
   {
      openLogFile();

      int blockIndex = 0;
      int length = (int) file.length();
      m_buffer = new byte[length];
      FileInputStream is = new FileInputStream(file);
      try
      {
         int bytesRead = is.read(m_buffer);
         if (bytesRead != length)
         {
            throw new RuntimeException("Read count different");
         }
      }
      finally
      {
         is.close();
      }

      List<Integer> blocks = new ArrayList<Integer>();
      for (int index = 64; index < m_buffer.length - 11; index++)
      {
         if (matchPattern(PARENT_BLOCK_PATTERNS, index))
         {
            blocks.add(Integer.valueOf(index));
         }
      }

      int startIndex = 0;
      for (int endIndex : blocks)
      {
         int blockLength = endIndex - startIndex;
         readBlock(blockIndex, startIndex, blockLength);
         startIndex = endIndex;
         ++blockIndex;
      }

      int blockLength = m_buffer.length - startIndex;
      readBlock(blockIndex, startIndex, blockLength);

      closeLogFile();
   }
935346
@Override
	public String read() throws Exception {
		String item = null;
		if (index < input.length) {
			item = input[index++];
			LOGGER.info(item);
			return item;
		} else {
			return null;
		}

	}
419444
public void readConfig(Reader reader) throws IOException {
		try {
			XMLStreamReader xr = XMLInputFactory.newInstance().createXMLStreamReader(reader);
			try {
				while (!xr.isStartElement()) {
					xr.next();
				}
				processStartElement(xr, "simon-configuration");
				while (true) {
					if (isStartTag(xr, "callback")) {
						manager.callback().addCallback(processCallback(xr));
					} else if (isStartTag(xr, "filter-callback")) {
						manager.callback().addCallback(processFilterCallback(xr));
					} else if (isStartTag(xr, "simon")) {
						processSimon(xr);
					} else {
						break;
					}
				}
				assertEndTag(xr, "simon-configuration");
			} finally {
				xr.close();
			}
		} catch (XMLStreamException e) {
			manager.callback().onManagerWarning(null, e);
		} catch (SimonException e) {
			manager.callback().onManagerWarning(e.getMessage(), e);
		}
	}
1013521
public static BufferedReader getReader(Reader reader) {
		if (null == reader) {
			return null;
		}

		return (reader instanceof BufferedReader) ? (BufferedReader) reader : new BufferedReader(reader);
	}
626868
@GwtIncompatible("Reader")
  static CharInput asCharInput(final Reader reader) {
    checkNotNull(reader);
    return new CharInput() {
      @Override
      public int read() throws IOException {
        return reader.read();
      }

      @Override
      public void close() throws IOException {
        reader.close();
      }
    };
  }
653953
public static <T extends Collection<String>> T readLines(Reader reader, final T collection) throws IORuntimeException {
		readLines(reader, new LineHandler() {
			@Override
			public void handle(String line) {
				collection.add(line);
			}
		});
		return collection;
	}
593970
public void parseReader(Reader reader) throws ParserException, IOException {

        try {
            nanoElement.parseFromReader(reader);
        } catch (XMLParseException e) {
            throw new ParserException(e.getMessage());
        }
    }
1034323
@SuppressWarnings("unchecked")
	private void initInputReaders() throws Exception {
		int numGates = 0;
		//  ---------------- create the input readers ---------------------
		// in case where a logical input unions multiple physical inputs, create a union reader
		final int groupSize = this.config.getGroupSize(0);
		numGates += groupSize;
		if (groupSize == 1) {
			// non-union case
			inputReader = new MutableRecordReader<DeserializationDelegate<IT>>(
					getEnvironment().getInputGate(0),
					getEnvironment().getTaskManagerInfo().getTmpDirectories());
		} else if (groupSize > 1){
			// union case
			inputReader = new MutableRecordReader<IOReadableWritable>(
					new UnionInputGate(getEnvironment().getAllInputGates()),
					getEnvironment().getTaskManagerInfo().getTmpDirectories());
		} else {
			throw new Exception("Illegal input group size in task configuration: " + groupSize);
		}
		
		this.inputTypeSerializerFactory = this.config.getInputSerializer(0, getUserCodeClassLoader());
		@SuppressWarnings({ "rawtypes" })
		final MutableObjectIterator<?> iter = new ReaderIterator(inputReader, this.inputTypeSerializerFactory.getSerializer());
		this.reader = (MutableObjectIterator<IT>)iter;

		// final sanity check
		if (numGates != this.config.getNumInputs()) {
			throw new Exception("Illegal configuration: Number of input gates and group sizes are not consistent.");
		}
	}
824168
synchronized void receive(int c) throws IOException {
        if (!connected) {
            throw new IOException("Pipe not connected");
        } else if (closedByWriter || closedByReader) {
            throw new IOException("Pipe closed");
        } else if (readSide != null && !readSide.isAlive()) {
            throw new IOException("Read end dead");
        }

        writeSide = Thread.currentThread();
        while (in == out) {
            if ((readSide != null) && !readSide.isAlive()) {
                throw new IOException("Pipe broken");
            }
            /* full: kick any waiting readers */
            notifyAll();
            try {
                wait(1000);
            } catch (InterruptedException ex) {
                throw new java.io.InterruptedIOException();
            }
        }
        if (in < 0) {
            in = 0;
            out = 0;
        }
        buffer[in++] = (char) c;
        if (in >= buffer.length) {
            in = 0;
        }
    }
584822
public static void loadFrom(Reader reader, Proto target) throws Exception
    {
        loadFrom(new ANTLRReaderStream(reader), target);
    }
1003643
public static <E> void pipe(Iterator<E> iterator, OutputIterator<E> outputIterator) {
        new ConsumeIntoOutputIterator<>(outputIterator).apply(iterator);
    }
781029
public void parse (Reader reader) {
		try {
			char[] data = new char[1024];
			int offset = 0;
			while (true) {
				int length = reader.read(data, offset, data.length - offset);
				if (length == -1) break;
				if (length == 0) {
					char[] newData = new char[data.length * 2];
					System.arraycopy(data, 0, newData, 0, data.length);
					data = newData;
				} else
					offset += length;
			}
			parse(data, 0, offset);
		} catch (IOException ex) {
			throw new SerializationException(ex);
		} finally {
			StreamUtils.closeQuietly(reader);
		}
	}
507649
public void runPipelineWithHashtag() throws Exception {

		DynoJedisPipeline pipeline = client.pipelined();
		try {

			pipeline.set("pipeline-hashtag1-{bar}", "value-1");
			pipeline.set("pipeline-hashtag2-{bar}", "value-2");
			pipeline.set("pipeline-hashtag3-{bar}", "value-3");
			pipeline.set("pipeline-hashtag4-{bar}", "value-4");
			pipeline.set("pipeline-hashtag5-{bar}", "value-5");
			pipeline.set("pipeline-hashtag6-{bar}", "value-6");
			pipeline.set("pipeline-hashtag7-{bar}", "value-7");
			pipeline.set("pipeline-hashtag8-{bar}", "value-8");
			pipeline.set("pipeline-hashtag9-{bar}", "value-9");
			pipeline.set("pipeline-hashtag10-{bar}", "value-10");

			Response<String> value1 = pipeline.get("pipeline-hashtag1-{bar}");
			Response<String> value2 = pipeline.get("pipeline-hashtag2-{bar}");
			Response<String> value3 = pipeline.get("pipeline-hashtag3-{bar}");
			Response<String> value4 = pipeline.get("pipeline-hashtag4-{bar}");
			Response<String> value5 = pipeline.get("pipeline-hashtag5-{bar}");
			Response<String> value6 = pipeline.get("pipeline-hashtag6-{bar}");
			pipeline.sync();

			System.out.println(value1.get());
			System.out.println(value2.get());
			System.out.println(value3.get());
			System.out.println(value4.get());
			System.out.println(value5.get());
			System.out.println(value6.get());
		} catch (Exception e) {
			pipeline.discardPipelineAndReleaseConnection();
			throw e;
		}
	}
77104
public static ReadStreamOld openRead(Reader reader)
  {
    if (reader instanceof ReadStreamOld.StreamReader)
      return ((ReadStreamOld.StreamReader) reader).getStream();
    
    ReaderWriterStream s = new ReaderWriterStream(reader, null);
    ReadStreamOld is = new ReadStreamOld(s);
    try {
      is.setEncoding("utf-8");
    } catch (Exception e) {
    }

    return is;
  }
951846
private static void process(String input, String output) throws Exception
   {
      FileInputStream is = new FileInputStream(input);
      PrintWriter pw = new PrintWriter(new FileWriter(output));

      POIFSFileSystem fs = new POIFSFileSystem(is);
      dumpTree(pw, fs.getRoot(), "", true, true, null);

      is.close();
      pw.flush();
      pw.close();
      fs.close();
   }