update

hdata-core/src/main/java/com/github/stuxuhai/hdata/CliDriver.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata;
 
 import java.util.Map.Entry;
@@ -31,134 +24,134 @@ import com.google.common.base.Throwables;
 
 public class CliDriver {
 
-  private static final String XML_FILE_OPTION = "f";
-  private static final String HDATA_VARS_OPTION = "D";
-  private static final String QUIET_OPTION = "q";
-  private static final String READER_OPTION = "reader";
-  private static final String WRITER_OPTION = "writer";
-  private static final String READER_VARS_OPTION = "R";
-  private static final String WRITER_VARS_OPTION = "W";
-
-  private static final Logger LOGGER = LogManager.getLogger();
-
-  /**
-   * 创建命令行选项
-   * 
-   * @return
-   */
-  public Options createOptions() {
-    Options options = new Options();
-    options.addOption(XML_FILE_OPTION, null, true, "job xml path");
-    options.addOption(QUIET_OPTION, null, false, "quiet");
-    options.addOption(Option.builder(HDATA_VARS_OPTION).hasArgs().build());
-
-    options.addOption(null, READER_OPTION, true, "reader name");
-    options.addOption(Option.builder(READER_VARS_OPTION).hasArgs().build());
-
-    options.addOption(null, WRITER_OPTION, true, "writer name");
-    options.addOption(Option.builder(WRITER_VARS_OPTION).hasArgs().build());
-    return options;
-  }
-
-  /**
-   * 打印命令行帮助信息
-   * 
-   * @param options
-   */
-  public void printHelp(Options options) {
-    HelpFormatter formatter = new HelpFormatter();
-    formatter.printHelp(" ", options);
-  }
-
-  /**
-   * 替换命令行变量
-   * 
-   * @param config
-   * @param vars
-   */
-  public void replaceConfigVars(PluginConfig config, Properties vars) {
-    for (Entry<Object, Object> confEntry : config.entrySet()) {
-      if (confEntry.getKey().getClass() == String.class && confEntry.getValue().getClass() == String.class) {
-        for (Entry<Object, Object> varEntry : vars.entrySet()) {
-          String replaceVar = "${" + varEntry.getKey() + "}";
-          if (confEntry.getValue().toString().contains(replaceVar)) {
-            config.put(confEntry.getKey(), confEntry.getValue().toString().replace(replaceVar, varEntry.getValue().toString()));
-          }
-        }
-      }
+    private static final String XML_FILE_OPTION = "f";
+    private static final String HDATA_VARS_OPTION = "D";
+    private static final String QUIET_OPTION = "q";
+    private static final String READER_OPTION = "reader";
+    private static final String WRITER_OPTION = "writer";
+    private static final String READER_VARS_OPTION = "R";
+    private static final String WRITER_VARS_OPTION = "W";
+
+    private static final Logger LOGGER = LogManager.getLogger();
+
+    /**
+     * 创建命令行选项
+     * 
+     * @return
+     */
+    public Options createOptions() {
+        Options options = new Options();
+        options.addOption(XML_FILE_OPTION, null, true, "job xml path");
+        options.addOption(QUIET_OPTION, null, false, "quiet");
+        options.addOption(Option.builder(HDATA_VARS_OPTION).hasArgs().build());
+
+        options.addOption(null, READER_OPTION, true, "reader name");
+        options.addOption(Option.builder(READER_VARS_OPTION).hasArgs().build());
+
+        options.addOption(null, WRITER_OPTION, true, "writer name");
+        options.addOption(Option.builder(WRITER_VARS_OPTION).hasArgs().build());
+        return options;
     }
-  }
 
-  private void putOptionValues(Properties props, String[] values) {
-    if (props != null && values != null) {
-      for (int i = 0, len = values.length; i < len; i++) {
-        props.put(values[i], values[++i]);
-      }
-    }
-  }
-
-  /**
-   * 主程序入口
-   * 
-   * @param args
-   */
-  public static void main(String[] args) {
-
-    CliDriver cliDriver = new CliDriver();
-    Options options = cliDriver.createOptions();
-    if (args.length < 1) {
-      cliDriver.printHelp(options);
-      System.exit(-1);
+    /**
+     * 打印命令行帮助信息
+     * 
+     * @param options
+     */
+    public void printHelp(Options options) {
+        HelpFormatter formatter = new HelpFormatter();
+        formatter.printHelp(" ", options);
     }
 
-    CommandLineParser parser = new DefaultParser();
-    CommandLine cmd = null;
-    try {
-      cmd = parser.parse(options, args);
-      if (cmd.hasOption(QUIET_OPTION)) {
-        LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
-        Configuration conf = ctx.getConfiguration();
-        conf.getLoggerConfig(LogManager.ROOT_LOGGER_NAME).setLevel(Level.WARN);
-        ctx.updateLoggers(conf);
-      }
-
-      final DefaultJobConfig jobConfig;
-      if (cmd.hasOption(XML_FILE_OPTION)) {
-        String jobXmlPath = cmd.getOptionValue(XML_FILE_OPTION);
-        jobConfig = DefaultJobConfig.createFromXML(jobXmlPath);
-        Properties vars = new Properties();
-        cliDriver.putOptionValues(vars, cmd.getOptionValues(HDATA_VARS_OPTION));
-
-        final PluginConfig readerConfig = jobConfig.getReaderConfig();
-        final PluginConfig writerConfig = jobConfig.getWriterConfig();
-
-        cliDriver.replaceConfigVars(readerConfig, vars);
-        cliDriver.replaceConfigVars(writerConfig, vars);
-      } else {
-        if (!cmd.hasOption(READER_OPTION) || !cmd.hasOption(WRITER_OPTION)) {
-          throw new HDataException("Option --reader and --writer should be both given if -f option not exists.");
+    /**
+     * 替换命令行变量
+     * 
+     * @param config
+     * @param vars
+     */
+    public void replaceConfigVars(PluginConfig config, Properties vars) {
+        for (Entry<Object, Object> confEntry : config.entrySet()) {
+            if (confEntry.getKey().getClass() == String.class && confEntry.getValue().getClass() == String.class) {
+                for (Entry<Object, Object> varEntry : vars.entrySet()) {
+                    String replaceVar = "${" + varEntry.getKey() + "}";
+                    if (confEntry.getValue().toString().contains(replaceVar)) {
+                        config.put(confEntry.getKey(), confEntry.getValue().toString().replace(replaceVar, varEntry.getValue().toString()));
+                    }
+                }
+            }
         }
+    }
 
-        String readerName = cmd.getOptionValue(READER_OPTION);
-        String writerName = cmd.getOptionValue(WRITER_OPTION);
-
-        PluginConfig readerConfig = new PluginConfig();
-        cliDriver.putOptionValues(readerConfig, cmd.getOptionValues(READER_VARS_OPTION));
-
-        PluginConfig writerConfig = new PluginConfig();
-        cliDriver.putOptionValues(writerConfig, cmd.getOptionValues(WRITER_VARS_OPTION));
+    private void putOptionValues(Properties props, String[] values) {
+        if (props != null && values != null) {
+            for (int i = 0, len = values.length; i < len; i++) {
+                props.put(values[i], values[++i]);
+            }
+        }
+    }
 
-        jobConfig = new DefaultJobConfig(readerName, readerConfig, writerName, writerConfig);
-      }
+    /**
+     * 主程序入口
+     * 
+     * @param args
+     */
+    public static void main(String[] args) {
+
+        CliDriver cliDriver = new CliDriver();
+        Options options = cliDriver.createOptions();
+        if (args.length < 1) {
+            cliDriver.printHelp(options);
+            System.exit(-1);
+        }
 
-      HData hData = new HData();
-      hData.start(jobConfig);
-    } catch (ParseException e) {
-      cliDriver.printHelp(options);
-      System.exit(-1);
-    } catch (Exception e) {
-      LOGGER.error(Throwables.getStackTraceAsString(e));
-      System.exit(-1);
+        CommandLineParser parser = new DefaultParser();
+        CommandLine cmd = null;
+        try {
+            cmd = parser.parse(options, args);
+            if (cmd.hasOption(QUIET_OPTION)) {
+                LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+                Configuration conf = ctx.getConfiguration();
+                conf.getLoggerConfig(LogManager.ROOT_LOGGER_NAME).setLevel(Level.WARN);
+                ctx.updateLoggers(conf);
+            }
+
+            final DefaultJobConfig jobConfig;
+            if (cmd.hasOption(XML_FILE_OPTION)) {
+                String jobXmlPath = cmd.getOptionValue(XML_FILE_OPTION);
+                jobConfig = DefaultJobConfig.createFromXML(jobXmlPath);
+                Properties vars = new Properties();
+                cliDriver.putOptionValues(vars, cmd.getOptionValues(HDATA_VARS_OPTION));
+
+                final PluginConfig readerConfig = jobConfig.getReaderConfig();
+                final PluginConfig writerConfig = jobConfig.getWriterConfig();
+
+                cliDriver.replaceConfigVars(readerConfig, vars);
+                cliDriver.replaceConfigVars(writerConfig, vars);
+            } else {
+                if (!cmd.hasOption(READER_OPTION) || !cmd.hasOption(WRITER_OPTION)) {
+                    throw new HDataException("Option --reader and --writer should be both given if -f option not exists.");
+                }
+
+                String readerName = cmd.getOptionValue(READER_OPTION);
+                String writerName = cmd.getOptionValue(WRITER_OPTION);
+
+                PluginConfig readerConfig = new PluginConfig();
+                cliDriver.putOptionValues(readerConfig, cmd.getOptionValues(READER_VARS_OPTION));
+
+                PluginConfig writerConfig = new PluginConfig();
+                cliDriver.putOptionValues(writerConfig, cmd.getOptionValues(WRITER_VARS_OPTION));
+
+                jobConfig = new DefaultJobConfig(readerName, readerConfig, writerName, writerConfig);
+            }
+
+            HData hData = new HData();
+            hData.start(jobConfig);
+        } catch (ParseException e) {
+            cliDriver.printHelp(options);
+            System.exit(-1);
+        } catch (Exception e) {
+            LOGGER.error(Throwables.getStackTraceAsString(e));
+            System.exit(-1);
+        }
     }
-  }
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/common/Constants.java

@@ -1,19 +1,12 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.common;
 
 public interface Constants {
 
-	public static final String HDATA_XML = "hdata.xml";
-	public static final String PLUGINS_XML = "plugins.xml";
-	public static final String LOG4J2_XML = "log4j2.xml";
-	public static final String DATE_FORMAT_STRING = "yyyy-MM-dd HH:mm:ss";
-	public static final String COLUMNS_SPLIT_REGEX = "\\s*,\\s*";
-	public static final long DEFAULT_HDATA_SLEEP_MILLIS = 3000;
+    public static final String HDATA_XML = "hdata.xml";
+    public static final String PLUGINS_XML = "plugins.xml";
+    public static final String LOG4J2_XML = "log4j2.xml";
+    public static final String DATE_FORMAT_STRING = "yyyy-MM-dd HH:mm:ss";
+    public static final String COLUMNS_SPLIT_REGEX = "\\s*,\\s*";
+    public static final long DEFAULT_HDATA_SLEEP_MILLIS = 3000;
 
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/common/HDataConfigConstants.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.common;
 
 public interface HDataConfigConstants {

hdata-core/src/main/java/com/github/stuxuhai/hdata/config/DefaultEngineConfig.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.config;
 
 import java.util.List;
@@ -20,31 +13,31 @@ import com.google.common.base.Throwables;
 
 public class DefaultEngineConfig extends EngineConfig {
 
-	private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = 1L;
 
-	private DefaultEngineConfig() {
-		super();
-	}
+    private DefaultEngineConfig() {
+        super();
+    }
 
-	public static DefaultEngineConfig create() {
-		DefaultEngineConfig conf = new DefaultEngineConfig();
-		String path = Utils.getConfigDir() + Constants.HDATA_XML;
+    public static DefaultEngineConfig create() {
+        DefaultEngineConfig conf = new DefaultEngineConfig();
+        String path = Utils.getConfigDir() + Constants.HDATA_XML;
 
-		try {
-			XMLConfiguration config = new XMLConfiguration(path);
-			config.setValidating(true);
+        try {
+            XMLConfiguration config = new XMLConfiguration(path);
+            config.setValidating(true);
 
-			List<HierarchicalConfiguration> properties = config.configurationsAt(".property");
-			for (HierarchicalConfiguration hc : properties) {
-				String name = hc.getString("name");
-				String value = hc.getString("value");
-				conf.setProperty(name, value);
-			}
-		} catch (ConfigurationException e) {
-			Throwables.propagate(e);
-		}
+            List<HierarchicalConfiguration> properties = config.configurationsAt(".property");
+            for (HierarchicalConfiguration hc : properties) {
+                String name = hc.getString("name");
+                String value = hc.getString("value");
+                conf.setProperty(name, value);
+            }
+        } catch (ConfigurationException e) {
+            Throwables.propagate(e);
+        }
 
-		return conf;
-	}
+        return conf;
+    }
 
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/config/DefaultJobConfig.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.config;
 
 import java.util.Iterator;
@@ -26,103 +19,102 @@ import com.google.common.base.Throwables;
 
 public class DefaultJobConfig extends JobConfig {
 
-	private final PluginConfig readerConfig;
-	private final PluginConfig writerConfig;
-	private final String readerName;
-	private final String writerName;
-	private static final long serialVersionUID = 1L;
-
-	public DefaultJobConfig(String readerName, PluginConfig readerConfig, String writerName,
-			PluginConfig writerConfig) {
-		super();
-		this.readerName = readerName;
-		this.readerConfig = readerConfig;
-		this.writerName = writerName;
-		this.writerConfig = writerConfig;
-	}
-
-	@Override
-	public PluginConfig getReaderConfig() {
-		return readerConfig;
-	}
-
-	@Override
-	public PluginConfig getWriterConfig() {
-		return writerConfig;
-	}
-
-	@Override
-	public String getReaderName() {
-		return readerName;
-	}
-
-	@Override
-	public String getWriterName() {
-		return writerName;
-	}
-
-	@Override
-	public Reader newReader() {
-		String readerClassName = PluginLoader.getReaderClassName(readerName);
-		Preconditions.checkNotNull(readerClassName, "Can not find class for reader: " + readerName);
-
-		try {
-			return (Reader) PluginUtils.loadClass(readerName, readerClassName).newInstance();
-		} catch (Exception e) {
-			throw new HDataException("Can not create new reader instance for: " + readerName, e);
-		}
-	}
-
-	@Override
-	public Splitter newSplitter() {
-		Reader reader = newReader();
-		return reader.newSplitter();
-	}
-
-	@Override
-	public Writer newWriter() {
-		String writerClassName = PluginLoader.getWriterClassName(writerName);
-		Preconditions.checkNotNull(writerClassName, "Can not find class for writer: " + writerName);
-
-		try {
-			return (Writer) PluginUtils.loadClass(writerName, writerClassName).newInstance();
-		} catch (Exception e) {
-			throw new HDataException("Can not create new writer instance for: " + writerName, e);
-		}
-	}
-
-	public static DefaultJobConfig createFromXML(String path) {
-		try {
-			XMLConfiguration xmlConfig = new XMLConfiguration(path);
-			xmlConfig.setValidating(true);
-
-			PluginConfig readerPluginConfig = new PluginConfig();
-			String readerName = xmlConfig.getString("reader[@name]");
-			SubnodeConfiguration readerSc = xmlConfig.configurationAt("reader");
-			Iterator<String> readerIt = readerSc.getKeys();
-			while (readerIt.hasNext()) {
-				String key = readerIt.next();
-				if (!key.startsWith("[@")) {
-					readerPluginConfig.setProperty(key.replace("..", "."), readerSc.getString(key));
-				}
-			}
-
-			PluginConfig writerPluginConfig = new PluginConfig();
-			String writerName = xmlConfig.getString("writer[@name]");
-			SubnodeConfiguration writerSc = xmlConfig.configurationAt("writer");
-			Iterator<String> writerIt = writerSc.getKeys();
-			while (writerIt.hasNext()) {
-				String key = writerIt.next();
-				if (!key.startsWith("[@")) {
-					writerPluginConfig.setProperty(key.replace("..", "."), writerSc.getString(key));
-				}
-			}
-
-			return new DefaultJobConfig(readerName, readerPluginConfig, writerName, writerPluginConfig);
-		} catch (ConfigurationException e) {
-			Throwables.propagate(e);
-		}
-
-		return null;
-	}
+    private final PluginConfig readerConfig;
+    private final PluginConfig writerConfig;
+    private final String readerName;
+    private final String writerName;
+    private static final long serialVersionUID = 1L;
+
+    public DefaultJobConfig(String readerName, PluginConfig readerConfig, String writerName, PluginConfig writerConfig) {
+        super();
+        this.readerName = readerName;
+        this.readerConfig = readerConfig;
+        this.writerName = writerName;
+        this.writerConfig = writerConfig;
+    }
+
+    @Override
+    public PluginConfig getReaderConfig() {
+        return readerConfig;
+    }
+
+    @Override
+    public PluginConfig getWriterConfig() {
+        return writerConfig;
+    }
+
+    @Override
+    public String getReaderName() {
+        return readerName;
+    }
+
+    @Override
+    public String getWriterName() {
+        return writerName;
+    }
+
+    @Override
+    public Reader newReader() {
+        String readerClassName = PluginLoader.getReaderClassName(readerName);
+        Preconditions.checkNotNull(readerClassName, "Can not find class for reader: " + readerName);
+
+        try {
+            return (Reader) PluginUtils.loadClass(readerName, readerClassName).newInstance();
+        } catch (Exception e) {
+            throw new HDataException("Can not create new reader instance for: " + readerName, e);
+        }
+    }
+
+    @Override
+    public Splitter newSplitter() {
+        Reader reader = newReader();
+        return reader.newSplitter();
+    }
+
+    @Override
+    public Writer newWriter() {
+        String writerClassName = PluginLoader.getWriterClassName(writerName);
+        Preconditions.checkNotNull(writerClassName, "Can not find class for writer: " + writerName);
+
+        try {
+            return (Writer) PluginUtils.loadClass(writerName, writerClassName).newInstance();
+        } catch (Exception e) {
+            throw new HDataException("Can not create new writer instance for: " + writerName, e);
+        }
+    }
+
+    public static DefaultJobConfig createFromXML(String path) {
+        try {
+            XMLConfiguration xmlConfig = new XMLConfiguration(path);
+            xmlConfig.setValidating(true);
+
+            PluginConfig readerPluginConfig = new PluginConfig();
+            String readerName = xmlConfig.getString("reader[@name]");
+            SubnodeConfiguration readerSc = xmlConfig.configurationAt("reader");
+            Iterator<String> readerIt = readerSc.getKeys();
+            while (readerIt.hasNext()) {
+                String key = readerIt.next();
+                if (!key.startsWith("[@")) {
+                    readerPluginConfig.setProperty(key.replace("..", "."), readerSc.getString(key));
+                }
+            }
+
+            PluginConfig writerPluginConfig = new PluginConfig();
+            String writerName = xmlConfig.getString("writer[@name]");
+            SubnodeConfiguration writerSc = xmlConfig.configurationAt("writer");
+            Iterator<String> writerIt = writerSc.getKeys();
+            while (writerIt.hasNext()) {
+                String key = writerIt.next();
+                if (!key.startsWith("[@")) {
+                    writerPluginConfig.setProperty(key.replace("..", "."), writerSc.getString(key));
+                }
+            }
+
+            return new DefaultJobConfig(readerName, readerPluginConfig, writerName, writerPluginConfig);
+        } catch (ConfigurationException e) {
+            Throwables.propagate(e);
+        }
+
+        return null;
+    }
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/DefaultRecord.java

@@ -1,41 +1,34 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.core;
 
 import com.github.stuxuhai.hdata.api.Record;
 
 public class DefaultRecord implements Record {
 
-	private final Object[] fields;
-	private int cursor;
-
-	public DefaultRecord(int fieldCount) {
-		fields = new Object[fieldCount];
-	}
-
-	@Override
-	public void add(int index, Object field) {
-		fields[index] = field;
-		this.cursor++;
-	}
-
-	@Override
-	public void add(Object field) {
-		add(cursor, field);
-	}
-
-	@Override
-	public Object get(int index) {
-		return fields[index];
-	}
-
-	@Override
-	public int size() {
-		return fields.length;
-	}
+    private final Object[] fields;
+    private int cursor;
+
+    public DefaultRecord(int fieldCount) {
+        fields = new Object[fieldCount];
+    }
+
+    @Override
+    public void add(int index, Object field) {
+        fields[index] = field;
+        this.cursor++;
+    }
+
+    @Override
+    public void add(Object field) {
+        add(cursor, field);
+    }
+
+    @Override
+    public Object get(int index) {
+        return fields[index];
+    }
+
+    @Override
+    public int size() {
+        return fields.length;
+    }
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/DefaultRecordCollector.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.core;
 
 import java.util.concurrent.TimeUnit;
@@ -21,52 +14,51 @@ import com.google.common.base.Stopwatch;
 
 public class DefaultRecordCollector implements RecordCollector {
 
-	private static final Logger LOGGER = LogManager.getLogger(DefaultRecordCollector.class);
+    private static final Logger LOGGER = LogManager.getLogger(DefaultRecordCollector.class);
 
-	private static final long SLEEP_MILL_SECONDS = 1000;
+    private static final long SLEEP_MILL_SECONDS = 1000;
 
-	private final DefaultStorage storage;
-	private final Metric metric;
-	private final long flowLimit;
-	private Stopwatch stopwatch = Stopwatch.createStarted();
+    private final DefaultStorage storage;
+    private final Metric metric;
+    private final long flowLimit;
+    private Stopwatch stopwatch = Stopwatch.createStarted();
 
-	public DefaultRecordCollector(DefaultStorage storage, Metric metric, long flowLimit) {
-		this.storage = storage;
-		this.metric = metric;
-		this.flowLimit = flowLimit;
-		LOGGER.info("The flow limit is {} bytes/s.", this.flowLimit);
-	}
+    public DefaultRecordCollector(DefaultStorage storage, Metric metric, long flowLimit) {
+        this.storage = storage;
+        this.metric = metric;
+        this.flowLimit = flowLimit;
+        LOGGER.info("The flow limit is {} bytes/s.", this.flowLimit);
+    }
 
-	@Override
-	public void send(Record record) {
-		// 限速
-		if (flowLimit > 0) {
-			while (true) {
-				long currentSpeed = metric.getSpeed();
-				if (currentSpeed > flowLimit) {
-					if (stopwatch.elapsed(TimeUnit.SECONDS) >= 5) {
-						LOGGER.info("Current Speed is {} MB/s, sleeping...",
-								String.format("%.2f", (double) currentSpeed / 1024 / 1024));
-						stopwatch.reset();
-					}
-					Utils.sleep(SLEEP_MILL_SECONDS);
-				} else {
-					break;
-				}
-			}
-		}
+    @Override
+    public void send(Record record) {
+        // 限速
+        if (flowLimit > 0) {
+            while (true) {
+                long currentSpeed = metric.getSpeed();
+                if (currentSpeed > flowLimit) {
+                    if (stopwatch.elapsed(TimeUnit.SECONDS) >= 5) {
+                        LOGGER.info("Current Speed is {} MB/s, sleeping...", String.format("%.2f", (double) currentSpeed / 1024 / 1024));
+                        stopwatch.reset();
+                    }
+                    Utils.sleep(SLEEP_MILL_SECONDS);
+                } else {
+                    break;
+                }
+            }
+        }
 
-		storage.put(record);
-		metric.getReadCount().incrementAndGet();
+        storage.put(record);
+        metric.getReadCount().incrementAndGet();
 
-		if (flowLimit > 0) {
-			metric.getReadBytes().addAndGet(RamUsageEstimator.sizeOf(record));
-		}
+        if (flowLimit > 0) {
+            metric.getReadBytes().addAndGet(RamUsageEstimator.sizeOf(record));
+        }
 
-	}
+    }
 
-	@Override
-	public void send(Record[] records) {
-		storage.put(records);
-	}
+    @Override
+    public void send(Record[] records) {
+        storage.put(records);
+    }
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/DefaultStorage.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.core;
 
 import com.github.stuxuhai.hdata.api.JobContext;
@@ -16,48 +9,48 @@ import com.lmax.disruptor.dsl.Disruptor;
 
 public class DefaultStorage implements Storage {
 
-	private final Disruptor<RecordEvent> disruptor;
-	private final RingBuffer<RecordEvent> ringBuffer;
-
-	private static final EventTranslatorOneArg<RecordEvent, Record> TRANSLATOR = new EventTranslatorOneArg<RecordEvent, Record>() {
-
-		@Override
-		public void translateTo(RecordEvent event, long sequence, Record record) {
-			event.setRecord(record);
-		}
-	};
-
-	public DefaultStorage(Disruptor<RecordEvent> disruptor, RecordWorkHandler[] handlers, JobContext context) {
-		this.disruptor = disruptor;
-		disruptor.setDefaultExceptionHandler(new RecordEventExceptionHandler(disruptor, context));
-		disruptor.handleEventsWithWorkerPool(handlers);
-		ringBuffer = disruptor.start();
-	}
-
-	@Override
-	public void put(Record record) {
-		disruptor.publishEvent(TRANSLATOR, record);
-	}
-
-	@Override
-	public void put(Record[] records) {
-		for (Record record : records) {
-			put(record);
-		}
-	}
-
-	@Override
-	public boolean isEmpty() {
-		return ringBuffer.remainingCapacity() == ringBuffer.getBufferSize();
-	}
-
-	@Override
-	public int size() {
-		return ringBuffer.getBufferSize();
-	}
-
-	@Override
-	public void close() {
-		disruptor.shutdown();
-	}
+    private final Disruptor<RecordEvent> disruptor;
+    private final RingBuffer<RecordEvent> ringBuffer;
+
+    private static final EventTranslatorOneArg<RecordEvent, Record> TRANSLATOR = new EventTranslatorOneArg<RecordEvent, Record>() {
+
+        @Override
+        public void translateTo(RecordEvent event, long sequence, Record record) {
+            event.setRecord(record);
+        }
+    };
+
+    public DefaultStorage(Disruptor<RecordEvent> disruptor, RecordWorkHandler[] handlers, JobContext context) {
+        this.disruptor = disruptor;
+        disruptor.setDefaultExceptionHandler(new RecordEventExceptionHandler(disruptor, context));
+        disruptor.handleEventsWithWorkerPool(handlers);
+        ringBuffer = disruptor.start();
+    }
+
+    @Override
+    public void put(Record record) {
+        disruptor.publishEvent(TRANSLATOR, record);
+    }
+
+    @Override
+    public void put(Record[] records) {
+        for (Record record : records) {
+            put(record);
+        }
+    }
+
+    @Override
+    public boolean isEmpty() {
+        return ringBuffer.remainingCapacity() == ringBuffer.getBufferSize();
+    }
+
+    @Override
+    public int size() {
+        return ringBuffer.getBufferSize();
+    }
+
+    @Override
+    public void close() {
+        disruptor.shutdown();
+    }
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/HData.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.core;
 
 import java.text.DecimalFormat;
@@ -42,259 +35,255 @@ import com.lmax.disruptor.dsl.ProducerType;
 
 public class HData {
 
-	private int exitCode = 0;
-	private static final DecimalFormat decimalFormat = new DecimalFormat("#0.00");
-	private static final Logger LOGGER = LogManager.getLogger(HData.class);
-
-	public void start(final DefaultJobConfig jobConfig) {
-		final PluginConfig readerConfig = jobConfig.getReaderConfig();
-		final PluginConfig writerConfig = jobConfig.getWriterConfig();
-
-		String readerName = jobConfig.getReaderName();
-		String writerName = jobConfig.getWriterName();
-		LOGGER.info("Reader: {}, Writer: {}", readerName, writerName);
-
-		final JobContext context = new JobContext();
-		context.setJobConfig(jobConfig);
-
-		final Metric metric = new Metric();
-		context.setMetric(metric);
-
-		final OutputFieldsDeclarer outputFieldsDeclarer = new OutputFieldsDeclarer();
-		context.setDeclarer(outputFieldsDeclarer);
-
-		final DefaultEngineConfig engineConfig = DefaultEngineConfig.create();
-		context.setEngineConfig(engineConfig);
-
-		long sleepMillis = engineConfig.getLong(HDataConfigConstants.HDATA_SLEEP_MILLIS,
-				Constants.DEFAULT_HDATA_SLEEP_MILLIS);
-
-		List<PluginConfig> readerConfigList = null;
-		final Splitter splitter = jobConfig.newSplitter();
-		if (splitter != null) {
-			LOGGER.info("Executing splitter for reader.");
-
-			ExecutorService es = Executors.newCachedThreadPool();
-			Callable<List<PluginConfig>> callable = new Callable<List<PluginConfig>>() {
-				@Override
-				public List<PluginConfig> call() throws Exception {
-					Thread.currentThread().setContextClassLoader(splitter.getClass().getClassLoader());
-					return splitter.split(jobConfig);
-				}
-			};
-
-			Future<List<PluginConfig>> future = es.submit(callable);
-			es.shutdown();
-			try {
-				readerConfigList = future.get();
-			} catch (InterruptedException | ExecutionException e) {
-				LOGGER.error("", e);
-				System.exit(JobStatus.FAILED.getStatus());
-			}
-
-			if (readerConfigList.isEmpty()) {
-				System.exit(JobStatus.SUCCESS.getStatus());
-			}
-		} else {
-			if (readerConfig.getParallelism() > 1) {
-				LOGGER.warn("Can not find splitter, reader parallelism is set to 1.");
-			}
-			readerConfigList = new ArrayList<PluginConfig>();
-			readerConfigList.add(readerConfig);
-		}
-
-		Reader[] readers = new Reader[readerConfigList.size()];
-		for (int i = 0, len = readers.length; i < len; i++) {
-			readers[i] = jobConfig.newReader();
-		}
-
-		context.setReaders(readers);
-
-		int writerParallelism = writerConfig.getParallelism();
-		LOGGER.info("Reader parallelism: {}, Writer parallelism: {}", readers.length, writerParallelism);
-
-		final Writer[] writers = new Writer[writerParallelism];
-		final RecordWorkHandler[] handlers = new RecordWorkHandler[writerParallelism];
-		for (int i = 0; i < writerParallelism; i++) {
-			writers[i] = jobConfig.newWriter();
-			handlers[i] = new RecordWorkHandler(writers[i], context, writerConfig);
-		}
-		context.setWriters(writers);
-
-		int bufferSize = engineConfig.getInt(HDataConfigConstants.STORAGE_BUFFER_SIZE, 1024);
-		String WaitStrategyName = engineConfig.getString(HDataConfigConstants.HDATA_STORAGE_DISRUPTOR_WAIT_STRATEGY,
-				BlockingWaitStrategy.class.getName());
-
-		DefaultStorage storage = createStorage(bufferSize, WaitStrategyName, readers.length, handlers, context);
-		context.setStorage(storage);
-
-		LOGGER.info("Transfer data from reader to writer...");
-
-		DefaultRecordCollector rc = new DefaultRecordCollector(storage, metric, readerConfig.getFlowLimit());
-		ExecutorService es = Executors.newFixedThreadPool(readers.length);
-		CompletionService<Integer> cs = new ExecutorCompletionService<Integer>(es);
-		for (int i = 0, len = readerConfigList.size(); i < len; i++) {
-			ReaderWorker readerWorker = new ReaderWorker(readers[i], context, readerConfigList.get(i), rc);
-			cs.submit(readerWorker);
-		}
-		es.shutdown();
-
-		metric.setReaderStartTime(System.currentTimeMillis());
-		metric.setWriterStartTime(System.currentTimeMillis());
-		while (!es.isTerminated()) {
-			if (context.isWriterError()) {
-				LOGGER.info("Write error.");
-				LOGGER.info("Closing reader and writer.");
-				// storage.close();
-				closeReaders(readers);
-				closeWriters(writers);
-				LOGGER.info("Job run failed!");
-				System.exit(JobStatus.FAILED.getStatus());
-			}
-
-			Utils.sleep(sleepMillis);
-			LOGGER.info("Read: {}\tWrite: {}", metric.getReadCount().get(), metric.getWriteCount().get());
-		}
-
-		context.setReaderFinished(true);
-		metric.setReaderEndTime(System.currentTimeMillis());
-
-		while (!storage.isEmpty()) {
-			if (context.isWriterError()) {
-				LOGGER.info("Write error.");
-				closeWriters(writers);
-				LOGGER.info("Job run failed!");
-				System.exit(JobStatus.FAILED.getStatus());
-			}
-			Utils.sleep(sleepMillis);
-
-			LOGGER.info("Read Finished(total: {}), Write: {}", metric.getReadCount().get(),
-					metric.getWriteCount().get());
-		}
-
-		storage.close();
-		LOGGER.info("Read Finished(total: {}), Write Finished(total: {})", metric.getReadCount().get(),
-				metric.getWriteCount().get());
-
-		for (int i = 0, len = readers.length; i < len; i++) {
-			try {
-				Future<Integer> future = cs.take();
-				if (future == null) {
-					LOGGER.info("Read error.");
-					closeWriters(writers);
-					LOGGER.info("Job run failed!");
-					System.exit(1);
-				}
-
-				Integer result = future.get();
-				if (result == null) {
-					result = -1;
-				}
-
-				if (result != 0) {
-					LOGGER.info("Read error.");
-					closeWriters(writers);
-					LOGGER.info("Job run failed!");
-					System.exit(result);
-				}
-			} catch (Exception e) {
-				LOGGER.error(Throwables.getStackTraceAsString(e));
-				exitCode = 1;
-			}
-		}
-
-		metric.setWriterEndTime(System.currentTimeMillis());
-		if (!closeWriters(writers)) {
-			exitCode = 1;
-		}
-
-		context.setWriterFinished(true);
-
-		double readSeconds = (metric.getReaderEndTime() - metric.getReaderStartTime()) / 1000d;
-		double writeSeconds = (metric.getWriterEndTime() - metric.getWriterStartTime()) / 1000d;
-		String readSpeed = decimalFormat.format(metric.getReadCount().get() / readSeconds);
-		String writeSpeed = decimalFormat.format(metric.getWriteCount().get() / writeSeconds);
-		LOGGER.info("Read spent time: {}s, Write spent time: {}s", decimalFormat.format(readSeconds),
-				decimalFormat.format(writeSeconds));
-		LOGGER.info("Read records: {}/s, Write records: {}/s", readSpeed, writeSpeed);
-
-		// PluginUtils.closeURLClassLoader();
-
-		System.exit(exitCode);
-	}
-
-	private DefaultStorage createStorage(int bufferSize, String WaitStrategyName, int producerCount,
-			RecordWorkHandler[] handlers, JobContext context) {
-		WaitStrategy waitStrategy = WaitStrategyFactory.build(WaitStrategyName);
-		ProducerType producerType;
-		if (producerCount == 1) {
-			producerType = ProducerType.SINGLE;
-		} else {
-			producerType = ProducerType.MULTI;
-		}
-		Disruptor<RecordEvent> disruptor = new Disruptor<RecordEvent>(RecordEvent.FACTORY, bufferSize,
-				Executors.defaultThreadFactory(), producerType, waitStrategy);
-		DefaultStorage storage = new DefaultStorage(disruptor, handlers, context);
-		return storage;
-	}
-
-	private boolean closeReaders(final Reader[] readers) {
-		ExecutorService es = Executors.newCachedThreadPool();
-		Callable<Boolean> callable = new Callable<Boolean>() {
-			@Override
-			public Boolean call() throws Exception {
-				Thread.currentThread().setContextClassLoader(readers[0].getClass().getClassLoader());
-				try {
-					for (Reader reader : readers) {
-						reader.close();
-					}
-
-					return true;
-				} catch (Exception e) {
-					LOGGER.error(Throwables.getStackTraceAsString(e));
-				}
-				return false;
-			}
-		};
-
-		Future<Boolean> future = es.submit(callable);
-		es.shutdown();
-
-		try {
-			return future.get();
-		} catch (InterruptedException | ExecutionException e) {
-			LOGGER.error(Throwables.getStackTraceAsString(e));
-			return false;
-		}
-	}
-
-	private boolean closeWriters(final Writer[] writers) {
-		ExecutorService es = Executors.newCachedThreadPool();
-		Callable<Boolean> callable = new Callable<Boolean>() {
-			@Override
-			public Boolean call() throws Exception {
-				Thread.currentThread().setContextClassLoader(writers[0].getClass().getClassLoader());
-				try {
-					for (Writer writer : writers) {
-						writer.close();
-					}
-
-					return true;
-				} catch (Exception e) {
-					LOGGER.error(Throwables.getStackTraceAsString(e));
-				}
-				return false;
-			}
-		};
-
-		Future<Boolean> future = es.submit(callable);
-		es.shutdown();
-
-		try {
-			return future.get();
-		} catch (InterruptedException | ExecutionException e) {
-			LOGGER.error(Throwables.getStackTraceAsString(e));
-			return false;
-		}
-	}
+    private int exitCode = 0;
+    private static final DecimalFormat decimalFormat = new DecimalFormat("#0.00");
+    private static final Logger LOGGER = LogManager.getLogger(HData.class);
+
+    public void start(final DefaultJobConfig jobConfig) {
+        final PluginConfig readerConfig = jobConfig.getReaderConfig();
+        final PluginConfig writerConfig = jobConfig.getWriterConfig();
+
+        String readerName = jobConfig.getReaderName();
+        String writerName = jobConfig.getWriterName();
+        LOGGER.info("Reader: {}, Writer: {}", readerName, writerName);
+
+        final JobContext context = new JobContext();
+        context.setJobConfig(jobConfig);
+
+        final Metric metric = new Metric();
+        context.setMetric(metric);
+
+        final OutputFieldsDeclarer outputFieldsDeclarer = new OutputFieldsDeclarer();
+        context.setDeclarer(outputFieldsDeclarer);
+
+        final DefaultEngineConfig engineConfig = DefaultEngineConfig.create();
+        context.setEngineConfig(engineConfig);
+
+        long sleepMillis = engineConfig.getLong(HDataConfigConstants.HDATA_SLEEP_MILLIS, Constants.DEFAULT_HDATA_SLEEP_MILLIS);
+
+        List<PluginConfig> readerConfigList = null;
+        final Splitter splitter = jobConfig.newSplitter();
+        if (splitter != null) {
+            LOGGER.info("Executing splitter for reader.");
+
+            ExecutorService es = Executors.newCachedThreadPool();
+            Callable<List<PluginConfig>> callable = new Callable<List<PluginConfig>>() {
+                @Override
+                public List<PluginConfig> call() throws Exception {
+                    Thread.currentThread().setContextClassLoader(splitter.getClass().getClassLoader());
+                    return splitter.split(jobConfig);
+                }
+            };
+
+            Future<List<PluginConfig>> future = es.submit(callable);
+            es.shutdown();
+            try {
+                readerConfigList = future.get();
+            } catch (InterruptedException | ExecutionException e) {
+                LOGGER.error("", e);
+                System.exit(JobStatus.FAILED.getStatus());
+            }
+
+            if (readerConfigList.isEmpty()) {
+                System.exit(JobStatus.SUCCESS.getStatus());
+            }
+        } else {
+            if (readerConfig.getParallelism() > 1) {
+                LOGGER.warn("Can not find splitter, reader parallelism is set to 1.");
+            }
+            readerConfigList = new ArrayList<PluginConfig>();
+            readerConfigList.add(readerConfig);
+        }
+
+        Reader[] readers = new Reader[readerConfigList.size()];
+        for (int i = 0, len = readers.length; i < len; i++) {
+            readers[i] = jobConfig.newReader();
+        }
+
+        context.setReaders(readers);
+
+        int writerParallelism = writerConfig.getParallelism();
+        LOGGER.info("Reader parallelism: {}, Writer parallelism: {}", readers.length, writerParallelism);
+
+        final Writer[] writers = new Writer[writerParallelism];
+        final RecordWorkHandler[] handlers = new RecordWorkHandler[writerParallelism];
+        for (int i = 0; i < writerParallelism; i++) {
+            writers[i] = jobConfig.newWriter();
+            handlers[i] = new RecordWorkHandler(writers[i], context, writerConfig);
+        }
+        context.setWriters(writers);
+
+        int bufferSize = engineConfig.getInt(HDataConfigConstants.STORAGE_BUFFER_SIZE, 1024);
+        String WaitStrategyName = engineConfig.getString(HDataConfigConstants.HDATA_STORAGE_DISRUPTOR_WAIT_STRATEGY,
+                BlockingWaitStrategy.class.getName());
+
+        DefaultStorage storage = createStorage(bufferSize, WaitStrategyName, readers.length, handlers, context);
+        context.setStorage(storage);
+
+        LOGGER.info("Transfer data from reader to writer...");
+
+        DefaultRecordCollector rc = new DefaultRecordCollector(storage, metric, readerConfig.getFlowLimit());
+        ExecutorService es = Executors.newFixedThreadPool(readers.length);
+        CompletionService<Integer> cs = new ExecutorCompletionService<Integer>(es);
+        for (int i = 0, len = readerConfigList.size(); i < len; i++) {
+            ReaderWorker readerWorker = new ReaderWorker(readers[i], context, readerConfigList.get(i), rc);
+            cs.submit(readerWorker);
+        }
+        es.shutdown();
+
+        metric.setReaderStartTime(System.currentTimeMillis());
+        metric.setWriterStartTime(System.currentTimeMillis());
+        while (!es.isTerminated()) {
+            if (context.isWriterError()) {
+                LOGGER.info("Write error.");
+                LOGGER.info("Closing reader and writer.");
+                // storage.close();
+                closeReaders(readers);
+                closeWriters(writers);
+                LOGGER.info("Job run failed!");
+                System.exit(JobStatus.FAILED.getStatus());
+            }
+
+            Utils.sleep(sleepMillis);
+            LOGGER.info("Read: {}\tWrite: {}", metric.getReadCount().get(), metric.getWriteCount().get());
+        }
+
+        context.setReaderFinished(true);
+        metric.setReaderEndTime(System.currentTimeMillis());
+
+        while (!storage.isEmpty()) {
+            if (context.isWriterError()) {
+                LOGGER.info("Write error.");
+                closeWriters(writers);
+                LOGGER.info("Job run failed!");
+                System.exit(JobStatus.FAILED.getStatus());
+            }
+            Utils.sleep(sleepMillis);
+
+            LOGGER.info("Read Finished(total: {}), Write: {}", metric.getReadCount().get(), metric.getWriteCount().get());
+        }
+
+        storage.close();
+        LOGGER.info("Read Finished(total: {}), Write Finished(total: {})", metric.getReadCount().get(), metric.getWriteCount().get());
+
+        for (int i = 0, len = readers.length; i < len; i++) {
+            try {
+                Future<Integer> future = cs.take();
+                if (future == null) {
+                    LOGGER.info("Read error.");
+                    closeWriters(writers);
+                    LOGGER.info("Job run failed!");
+                    System.exit(1);
+                }
+
+                Integer result = future.get();
+                if (result == null) {
+                    result = -1;
+                }
+
+                if (result != 0) {
+                    LOGGER.info("Read error.");
+                    closeWriters(writers);
+                    LOGGER.info("Job run failed!");
+                    System.exit(result);
+                }
+            } catch (Exception e) {
+                LOGGER.error(Throwables.getStackTraceAsString(e));
+                exitCode = 1;
+            }
+        }
+
+        metric.setWriterEndTime(System.currentTimeMillis());
+        if (!closeWriters(writers)) {
+            exitCode = 1;
+        }
+
+        context.setWriterFinished(true);
+
+        double readSeconds = (metric.getReaderEndTime() - metric.getReaderStartTime()) / 1000d;
+        double writeSeconds = (metric.getWriterEndTime() - metric.getWriterStartTime()) / 1000d;
+        String readSpeed = decimalFormat.format(metric.getReadCount().get() / readSeconds);
+        String writeSpeed = decimalFormat.format(metric.getWriteCount().get() / writeSeconds);
+        LOGGER.info("Read spent time: {}s, Write spent time: {}s", decimalFormat.format(readSeconds), decimalFormat.format(writeSeconds));
+        LOGGER.info("Read records: {}/s, Write records: {}/s", readSpeed, writeSpeed);
+
+        // PluginUtils.closeURLClassLoader();
+
+        System.exit(exitCode);
+    }
+
+    private DefaultStorage createStorage(int bufferSize, String WaitStrategyName, int producerCount, RecordWorkHandler[] handlers,
+            JobContext context) {
+        WaitStrategy waitStrategy = WaitStrategyFactory.build(WaitStrategyName);
+        ProducerType producerType;
+        if (producerCount == 1) {
+            producerType = ProducerType.SINGLE;
+        } else {
+            producerType = ProducerType.MULTI;
+        }
+        Disruptor<RecordEvent> disruptor = new Disruptor<RecordEvent>(RecordEvent.FACTORY, bufferSize, Executors.defaultThreadFactory(), producerType,
+                waitStrategy);
+        DefaultStorage storage = new DefaultStorage(disruptor, handlers, context);
+        return storage;
+    }
+
+    private boolean closeReaders(final Reader[] readers) {
+        ExecutorService es = Executors.newCachedThreadPool();
+        Callable<Boolean> callable = new Callable<Boolean>() {
+            @Override
+            public Boolean call() throws Exception {
+                Thread.currentThread().setContextClassLoader(readers[0].getClass().getClassLoader());
+                try {
+                    for (Reader reader : readers) {
+                        reader.close();
+                    }
+
+                    return true;
+                } catch (Exception e) {
+                    LOGGER.error(Throwables.getStackTraceAsString(e));
+                }
+                return false;
+            }
+        };
+
+        Future<Boolean> future = es.submit(callable);
+        es.shutdown();
+
+        try {
+            return future.get();
+        } catch (InterruptedException | ExecutionException e) {
+            LOGGER.error(Throwables.getStackTraceAsString(e));
+            return false;
+        }
+    }
+
+    private boolean closeWriters(final Writer[] writers) {
+        ExecutorService es = Executors.newCachedThreadPool();
+        Callable<Boolean> callable = new Callable<Boolean>() {
+            @Override
+            public Boolean call() throws Exception {
+                Thread.currentThread().setContextClassLoader(writers[0].getClass().getClassLoader());
+                try {
+                    for (Writer writer : writers) {
+                        writer.close();
+                    }
+
+                    return true;
+                } catch (Exception e) {
+                    LOGGER.error(Throwables.getStackTraceAsString(e));
+                }
+                return false;
+            }
+        };
+
+        Future<Boolean> future = es.submit(callable);
+        es.shutdown();
+
+        try {
+            return future.get();
+        } catch (InterruptedException | ExecutionException e) {
+            LOGGER.error(Throwables.getStackTraceAsString(e));
+            return false;
+        }
+    }
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/PluginClassLoader.java

@@ -1,11 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2016 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2016年4月10日 下午4:42:43
- */
-
 package com.github.stuxuhai.hdata.core;
 
 import java.net.URL;

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/PluginLoader.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.core;
 
 import java.util.HashMap;
@@ -21,41 +14,41 @@ import com.google.common.base.Throwables;
 
 public class PluginLoader {
 
-	private static Map<String, String> readerMap;
-	private static Map<String, String> writerMap;
-
-	public static String getReaderClassName(String name) {
-		return readerMap.get(name);
-	}
-
-	public static String getWriterClassName(String name) {
-		return writerMap.get(name);
-	}
-
-	static {
-		readerMap = new HashMap<String, String>();
-		writerMap = new HashMap<String, String>();
-
-		String path = Utils.getConfigDir() + Constants.PLUGINS_XML;
-		try {
-			XMLConfiguration config = new XMLConfiguration(path);
-			config.setValidating(true);
-
-			List<HierarchicalConfiguration> readerList = config.configurationsAt("readers.reader");
-			for (HierarchicalConfiguration hc : readerList) {
-				String name = hc.getString("name");
-				String clazz = hc.getString("class");
-				readerMap.put(name, clazz);
-			}
-
-			List<HierarchicalConfiguration> writerList = config.configurationsAt("writers.writer");
-			for (HierarchicalConfiguration hc : writerList) {
-				String name = hc.getString("name");
-				String clazz = hc.getString("class");
-				writerMap.put(name, clazz);
-			}
-		} catch (ConfigurationException e) {
-			Throwables.propagate(e);
-		}
-	}
+    private static Map<String, String> readerMap;
+    private static Map<String, String> writerMap;
+
+    public static String getReaderClassName(String name) {
+        return readerMap.get(name);
+    }
+
+    public static String getWriterClassName(String name) {
+        return writerMap.get(name);
+    }
+
+    static {
+        readerMap = new HashMap<String, String>();
+        writerMap = new HashMap<String, String>();
+
+        String path = Utils.getConfigDir() + Constants.PLUGINS_XML;
+        try {
+            XMLConfiguration config = new XMLConfiguration(path);
+            config.setValidating(true);
+
+            List<HierarchicalConfiguration> readerList = config.configurationsAt("readers.reader");
+            for (HierarchicalConfiguration hc : readerList) {
+                String name = hc.getString("name");
+                String clazz = hc.getString("class");
+                readerMap.put(name, clazz);
+            }
+
+            List<HierarchicalConfiguration> writerList = config.configurationsAt("writers.writer");
+            for (HierarchicalConfiguration hc : writerList) {
+                String name = hc.getString("name");
+                String clazz = hc.getString("class");
+                writerMap.put(name, clazz);
+            }
+        } catch (ConfigurationException e) {
+            Throwables.propagate(e);
+        }
+    }
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/ReaderWorker.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.core;
 
 import java.util.concurrent.Callable;
@@ -15,25 +8,25 @@ import com.github.stuxuhai.hdata.api.Reader;
 
 public class ReaderWorker implements Callable<Integer> {
 
-	private final Reader reader;
-	private final JobContext context;
-	private final DefaultRecordCollector rc;
-	private final PluginConfig readerConfig;
+    private final Reader reader;
+    private final JobContext context;
+    private final DefaultRecordCollector rc;
+    private final PluginConfig readerConfig;
 
-	public ReaderWorker(Reader reader, JobContext context, PluginConfig readerConfig, DefaultRecordCollector rc) {
-		this.reader = reader;
-		this.context = context;
-		this.rc = rc;
-		this.readerConfig = readerConfig;
-	}
+    public ReaderWorker(Reader reader, JobContext context, PluginConfig readerConfig, DefaultRecordCollector rc) {
+        this.reader = reader;
+        this.context = context;
+        this.rc = rc;
+        this.readerConfig = readerConfig;
+    }
 
-	@Override
-	public Integer call() throws Exception {
-		Thread.currentThread().setContextClassLoader(reader.getClass().getClassLoader());
-		reader.prepare(context, readerConfig);
-		reader.execute(rc);
-		reader.close();
-		return 0;
-	}
+    @Override
+    public Integer call() throws Exception {
+        Thread.currentThread().setContextClassLoader(reader.getClass().getClassLoader());
+        reader.prepare(context, readerConfig);
+        reader.execute(rc);
+        reader.close();
+        return 0;
+    }
 
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/RecordEvent.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.core;
 
 import com.github.stuxuhai.hdata.api.Record;
@@ -12,21 +5,21 @@ import com.lmax.disruptor.EventFactory;
 
 public class RecordEvent {
 
-	private Record record;
+    private Record record;
 
-	public Record getRecord() {
-		return record;
-	}
+    public Record getRecord() {
+        return record;
+    }
 
-	public void setRecord(Record record) {
-		this.record = record;
-	}
+    public void setRecord(Record record) {
+        this.record = record;
+    }
 
-	public static final EventFactory<RecordEvent> FACTORY = new EventFactory<RecordEvent>() {
+    public static final EventFactory<RecordEvent> FACTORY = new EventFactory<RecordEvent>() {
 
-		public RecordEvent newInstance() {
-			return new RecordEvent();
-		}
-	};
+        public RecordEvent newInstance() {
+            return new RecordEvent();
+        }
+    };
 
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/RecordEventExceptionHandler.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年7月2日 下午3:48:21
- */
 package com.github.stuxuhai.hdata.core;
 
 import org.apache.logging.log4j.LogManager;
@@ -15,10 +8,6 @@ import com.google.common.base.Throwables;
 import com.lmax.disruptor.ExceptionHandler;
 import com.lmax.disruptor.dsl.Disruptor;
 
-/**
- * @author wuya
- *
- */
 public class RecordEventExceptionHandler implements ExceptionHandler<Object> {
 
     private final Disruptor<RecordEvent> disruptor;

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/RecordWorkHandler.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.core;
 
 import com.github.stuxuhai.hdata.api.JobContext;
@@ -15,32 +8,32 @@ import com.lmax.disruptor.WorkHandler;
 
 public class RecordWorkHandler implements WorkHandler<RecordEvent> {
 
-	private final Writer writer;
-	private final JobContext context;
-	private final PluginConfig writerConfig;
-	private boolean writerPrepared;
-	private final Metric metric;
+    private final Writer writer;
+    private final JobContext context;
+    private final PluginConfig writerConfig;
+    private boolean writerPrepared;
+    private final Metric metric;
 
-	public RecordWorkHandler(Writer writer, JobContext context, PluginConfig writerConfig) {
-		this.writer = writer;
-		this.context = context;
-		this.writerConfig = writerConfig;
-		this.metric = context.getMetric();
-	}
+    public RecordWorkHandler(Writer writer, JobContext context, PluginConfig writerConfig) {
+        this.writer = writer;
+        this.context = context;
+        this.writerConfig = writerConfig;
+        this.metric = context.getMetric();
+    }
 
-	@Override
-	public void onEvent(RecordEvent event) {
-		if (!writerPrepared) {
-			context.declareOutputFields();
-			Thread.currentThread().setContextClassLoader(writer.getClass().getClassLoader());
-			writer.prepare(context, writerConfig);
-			writerPrepared = true;
-			if (metric.getWriterStartTime() == 0) {
-				metric.setWriterStartTime(System.currentTimeMillis());
-			}
-		}
+    @Override
+    public void onEvent(RecordEvent event) {
+        if (!writerPrepared) {
+            context.declareOutputFields();
+            Thread.currentThread().setContextClassLoader(writer.getClass().getClassLoader());
+            writer.prepare(context, writerConfig);
+            writerPrepared = true;
+            if (metric.getWriterStartTime() == 0) {
+                metric.setWriterStartTime(System.currentTimeMillis());
+            }
+        }
 
-		writer.execute(event.getRecord());
-		metric.getWriteCount().incrementAndGet();
-	}
+        writer.execute(event.getRecord());
+        metric.getWriteCount().incrementAndGet();
+    }
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/core/WaitStrategyFactory.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.core;
 
 import java.util.List;

hdata-core/src/main/java/com/github/stuxuhai/hdata/util/JdbcUtils.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.util;
 
 import java.sql.Connection;
@@ -25,187 +18,184 @@ import org.apache.commons.dbutils.ResultSetHandler;
 
 public class JdbcUtils {
 
-	/**
-	 * 获取表的字段类型
-	 *
-	 * @param connection
-	 * @param table
-	 * @return
-	 * @throws SQLException
-	 */
-	public static Map<String, Integer> getColumnTypes(Connection connection, String table) throws SQLException {
-		StringBuilder sql = new StringBuilder();
-		sql.append("SELECT * FROM ");
-		sql.append("`");
-		sql.append(table);
-		sql.append("`");
-		sql.append(" WHERE 1=2");
-		sql.append(" limit 1");
-
-		ResultSetHandler<Map<String, Integer>> handler = new ResultSetHandler<Map<String, Integer>>() {
-			@Override
-			public Map<String, Integer> handle(ResultSet rs) throws SQLException {
-				Map<String, Integer> map = new HashMap<String, Integer>();
-				ResultSetMetaData rsd = rs.getMetaData();
-				for (int i = 0; i < rsd.getColumnCount(); i++) {
-					map.put(rsd.getColumnName(i + 1).toLowerCase(), rsd.getColumnType(i + 1));
-				}
-				return map;
-			}
-		};
-
-		QueryRunner runner = new QueryRunner();
-		return runner.query(connection, sql.toString(), handler);
-	}
-
-	/**
-	 * 获取表的字段名称
-	 *
-	 * @param conn
-	 * @param table
-	 * @return
-	 * @throws SQLException
-	 */
-	public static List<String> getColumnNames(Connection conn, String table) throws SQLException {
-		StringBuilder sql = new StringBuilder();
-		sql.append("SELECT * FROM ");
-		sql.append("`");
-		sql.append(table);
-		sql.append("`");
-		sql.append(" WHERE 1=2");
-		sql.append(" limit 1");
-
-		ResultSetHandler<List<String>> handler = new ResultSetHandler<List<String>>() {
-
-			@Override
-			public List<String> handle(ResultSet rs) throws SQLException {
-				List<String> columnNames = new ArrayList<String>();
-				ResultSetMetaData rsd = rs.getMetaData();
-
-				for (int i = 0, len = rsd.getColumnCount(); i < len; i++) {
-					columnNames.add(rsd.getColumnName(i + 1));
-				}
-				return columnNames;
-			}
-		};
-
-		QueryRunner runner = new QueryRunner();
-		return runner.query(conn, sql.toString(), handler);
-	}
-
-	/**
-	 * 查询表中分割字段值的区域（最大值、最小值）
-	 *
-	 * @param conn
-	 * @param sql
-	 * @param splitColumn
-	 * @return
-	 * @throws SQLException
-	 */
-	public static double[] querySplitColumnRange(Connection conn, final String sql, final String splitColumn)
-			throws SQLException {
-		double[] minAndMax = new double[2];
-		Pattern p = Pattern.compile("\\s+FROM\\s+.*", Pattern.CASE_INSENSITIVE);
-		Matcher m = p.matcher(sql);
-
-		if (m.find() && splitColumn != null && !splitColumn.trim().isEmpty()) {
-			StringBuilder sb = new StringBuilder();
-			sb.append("SELECT MIN(");
-			sb.append(splitColumn);
-			sb.append("), MAX(");
-			sb.append(splitColumn);
-			sb.append(")");
-			sb.append(m.group(0));
-
-			ResultSetHandler<double[]> handler = new ResultSetHandler<double[]>() {
-
-				@Override
-				public double[] handle(ResultSet rs) throws SQLException {
-					double[] minAndMax = new double[2];
-					while (rs.next()) {
-						minAndMax[0] = rs.getDouble(1);
-						minAndMax[1] = rs.getDouble(2);
-					}
-
-					return minAndMax;
-				}
-			};
-
-			QueryRunner runner = new QueryRunner();
-			return runner.query(conn, sb.toString(), handler);
-		}
-
-		return minAndMax;
-	}
-
-	/**
-	 * 查询表数值类型的主键
-	 *
-	 * @param conn
-	 * @param catalog
-	 * @param schema
-	 * @param table
-	 * @return
-	 * @throws SQLException
-	 */
-	public static String getDigitalPrimaryKey(Connection conn, String catalog, String schema, String table)
-			throws SQLException {
-		List<String> primaryKeys = new ArrayList<String>();
-		ResultSet rs = conn.getMetaData().getPrimaryKeys(catalog, schema, table);
-		while (rs.next()) {
-			primaryKeys.add(rs.getString("COLUMN_NAME"));
-		}
-		rs.close();
-
-		if (primaryKeys.size() > 0) {
-			Map<String, Integer> map = getColumnTypes(conn, table);
-			for (String pk : primaryKeys) {
-				if (isDigitalType(map.get(pk.toLowerCase()))) {
-					return pk;
-				}
-			}
-		}
-
-		return null;
-	}
-
-	/**
-	 * 判断字段类型是否为数值类型
-	 *
-	 * @param sqlType
-	 * @return
-	 */
-	public static boolean isDigitalType(int sqlType) {
-		switch (sqlType) {
-		case Types.NUMERIC:
-		case Types.DECIMAL:
-		case Types.SMALLINT:
-		case Types.INTEGER:
-		case Types.BIGINT:
-		case Types.REAL:
-		case Types.FLOAT:
-		case Types.DOUBLE:
-			return true;
-
-		default:
-			return false;
-		}
-	}
-
-	public static boolean isStringType(int sqlType) {
-		switch (sqlType) {
-		case Types.CHAR:
-		case Types.VARCHAR:
-			return true;
-
-		default:
-			return false;
-		}
-	}
-
-	public static Connection getConnection(String driverClassName, String url, String username, String password)
-			throws Exception {
-		Class.forName(driverClassName);
-		return DriverManager.getConnection(url, username, password);
-	}
+    /**
+     * 获取表的字段类型
+     *
+     * @param connection
+     * @param table
+     * @return
+     * @throws SQLException
+     */
+    public static Map<String, Integer> getColumnTypes(Connection connection, String table) throws SQLException {
+        StringBuilder sql = new StringBuilder();
+        sql.append("SELECT * FROM ");
+        sql.append("`");
+        sql.append(table);
+        sql.append("`");
+        sql.append(" WHERE 1=2");
+        sql.append(" limit 1");
+
+        ResultSetHandler<Map<String, Integer>> handler = new ResultSetHandler<Map<String, Integer>>() {
+            @Override
+            public Map<String, Integer> handle(ResultSet rs) throws SQLException {
+                Map<String, Integer> map = new HashMap<String, Integer>();
+                ResultSetMetaData rsd = rs.getMetaData();
+                for (int i = 0; i < rsd.getColumnCount(); i++) {
+                    map.put(rsd.getColumnName(i + 1).toLowerCase(), rsd.getColumnType(i + 1));
+                }
+                return map;
+            }
+        };
+
+        QueryRunner runner = new QueryRunner();
+        return runner.query(connection, sql.toString(), handler);
+    }
+
+    /**
+     * 获取表的字段名称
+     *
+     * @param conn
+     * @param table
+     * @return
+     * @throws SQLException
+     */
+    public static List<String> getColumnNames(Connection conn, String table) throws SQLException {
+        StringBuilder sql = new StringBuilder();
+        sql.append("SELECT * FROM ");
+        sql.append("`");
+        sql.append(table);
+        sql.append("`");
+        sql.append(" WHERE 1=2");
+        sql.append(" limit 1");
+
+        ResultSetHandler<List<String>> handler = new ResultSetHandler<List<String>>() {
+
+            @Override
+            public List<String> handle(ResultSet rs) throws SQLException {
+                List<String> columnNames = new ArrayList<String>();
+                ResultSetMetaData rsd = rs.getMetaData();
+
+                for (int i = 0, len = rsd.getColumnCount(); i < len; i++) {
+                    columnNames.add(rsd.getColumnName(i + 1));
+                }
+                return columnNames;
+            }
+        };
+
+        QueryRunner runner = new QueryRunner();
+        return runner.query(conn, sql.toString(), handler);
+    }
+
+    /**
+     * 查询表中分割字段值的区域（最大值、最小值）
+     *
+     * @param conn
+     * @param sql
+     * @param splitColumn
+     * @return
+     * @throws SQLException
+     */
+    public static double[] querySplitColumnRange(Connection conn, final String sql, final String splitColumn) throws SQLException {
+        double[] minAndMax = new double[2];
+        Pattern p = Pattern.compile("\\s+FROM\\s+.*", Pattern.CASE_INSENSITIVE);
+        Matcher m = p.matcher(sql);
+
+        if (m.find() && splitColumn != null && !splitColumn.trim().isEmpty()) {
+            StringBuilder sb = new StringBuilder();
+            sb.append("SELECT MIN(");
+            sb.append(splitColumn);
+            sb.append("), MAX(");
+            sb.append(splitColumn);
+            sb.append(")");
+            sb.append(m.group(0));
+
+            ResultSetHandler<double[]> handler = new ResultSetHandler<double[]>() {
+
+                @Override
+                public double[] handle(ResultSet rs) throws SQLException {
+                    double[] minAndMax = new double[2];
+                    while (rs.next()) {
+                        minAndMax[0] = rs.getDouble(1);
+                        minAndMax[1] = rs.getDouble(2);
+                    }
+
+                    return minAndMax;
+                }
+            };
+
+            QueryRunner runner = new QueryRunner();
+            return runner.query(conn, sb.toString(), handler);
+        }
+
+        return minAndMax;
+    }
+
+    /**
+     * 查询表数值类型的主键
+     *
+     * @param conn
+     * @param catalog
+     * @param schema
+     * @param table
+     * @return
+     * @throws SQLException
+     */
+    public static String getDigitalPrimaryKey(Connection conn, String catalog, String schema, String table) throws SQLException {
+        List<String> primaryKeys = new ArrayList<String>();
+        ResultSet rs = conn.getMetaData().getPrimaryKeys(catalog, schema, table);
+        while (rs.next()) {
+            primaryKeys.add(rs.getString("COLUMN_NAME"));
+        }
+        rs.close();
+
+        if (primaryKeys.size() > 0) {
+            Map<String, Integer> map = getColumnTypes(conn, table);
+            for (String pk : primaryKeys) {
+                if (isDigitalType(map.get(pk.toLowerCase()))) {
+                    return pk;
+                }
+            }
+        }
+
+        return null;
+    }
+
+    /**
+     * 判断字段类型是否为数值类型
+     *
+     * @param sqlType
+     * @return
+     */
+    public static boolean isDigitalType(int sqlType) {
+        switch (sqlType) {
+        case Types.NUMERIC:
+        case Types.DECIMAL:
+        case Types.SMALLINT:
+        case Types.INTEGER:
+        case Types.BIGINT:
+        case Types.REAL:
+        case Types.FLOAT:
+        case Types.DOUBLE:
+            return true;
+
+        default:
+            return false;
+        }
+    }
+
+    public static boolean isStringType(int sqlType) {
+        switch (sqlType) {
+        case Types.CHAR:
+        case Types.VARCHAR:
+            return true;
+
+        default:
+            return false;
+        }
+    }
+
+    public static Connection getConnection(String driverClassName, String url, String username, String password) throws Exception {
+        Class.forName(driverClassName);
+        return DriverManager.getConnection(url, username, password);
+    }
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/util/NumberUtils.java

@@ -1,31 +1,24 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: xingtian
- * Create Date: 2015年1月16日 下午3:35:16
- */
 package com.github.stuxuhai.hdata.util;
 
 /**
  * 数字处理工具类
  *
- * */
+ */
 public class NumberUtils {
-	/**
-	 * 获取 起始和结束 范围内的所有 数值
-	 * 
-	 * */
-	public static int[] getRange(int before, int after) {
-		int bigger = Math.max(before, after);
-		int smaller = Math.min(before, after);
+    /**
+     * 获取 起始和结束 范围内的所有 数值
+     * 
+     */
+    public static int[] getRange(int before, int after) {
+        int bigger = Math.max(before, after);
+        int smaller = Math.min(before, after);
 
-		int[] range = new int[bigger + 1 - smaller];
-		for (int i = smaller; i <= bigger; i++) {
-			range[i - smaller] = i;
-		}
+        int[] range = new int[bigger + 1 - smaller];
+        for (int i = smaller; i <= bigger; i++) {
+            range[i - smaller] = i;
+        }
 
-		return range;
-	}
+        return range;
+    }
 
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/util/PluginUtils.java

@@ -1,11 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2016 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2016年4月9日 上午11:39:24
- */
-
 package com.github.stuxuhai.hdata.util;
 
 import java.io.File;
@@ -26,42 +18,41 @@ import com.google.common.collect.Maps;
 
 public class PluginUtils {
 
-	private static Map<String, PluginClassLoader> cache = Maps.newConcurrentMap();
-	private static final Logger LOGGER = LogManager.getLogger();
-
-	private static List<URL> listFileByPluginName(String pluginName) throws MalformedURLException {
-		List<URL> result = Lists.newArrayList();
-		File file = new File(PluginUtils.class.getProtectionDomain().getCodeSource().getLocation().getPath()
-				.replaceAll("/lib/.*\\.jar", "") + "/plugins/" + pluginName);
-		if (!file.exists()) {
-			throw new HDataException("Plugin not found: " + pluginName);
-		}
-
-		File[] jars = file.listFiles();
-		for (File jar : jars) {
-			result.add(jar.toURI().toURL());
-		}
-		return result;
-	}
-
-	public static Class<?> loadClass(String pluginName, String className)
-			throws ClassNotFoundException, MalformedURLException {
-		List<URL> list = listFileByPluginName(pluginName);
-		PluginClassLoader classLoader = cache.get(pluginName);
-		if (classLoader == null) {
-			classLoader = new PluginClassLoader(list.toArray(new URL[list.size()]), null);
-			cache.put(pluginName, classLoader);
-		}
-		return classLoader.loadClass(className);
-	}
-
-	public static void closeURLClassLoader() {
-		for (Entry<String, PluginClassLoader> entry : cache.entrySet()) {
-			try {
-				entry.getValue().close();
-			} catch (IOException e) {
-				LOGGER.error("", e);
-			}
-		}
-	}
+    private static Map<String, PluginClassLoader> cache = Maps.newConcurrentMap();
+    private static final Logger LOGGER = LogManager.getLogger();
+
+    private static List<URL> listFileByPluginName(String pluginName) throws MalformedURLException {
+        List<URL> result = Lists.newArrayList();
+        File file = new File(PluginUtils.class.getProtectionDomain().getCodeSource().getLocation().getPath().replaceAll("/lib/.*\\.jar", "")
+                + "/plugins/" + pluginName);
+        if (!file.exists()) {
+            throw new HDataException("Plugin not found: " + pluginName);
+        }
+
+        File[] jars = file.listFiles();
+        for (File jar : jars) {
+            result.add(jar.toURI().toURL());
+        }
+        return result;
+    }
+
+    public static Class<?> loadClass(String pluginName, String className) throws ClassNotFoundException, MalformedURLException {
+        List<URL> list = listFileByPluginName(pluginName);
+        PluginClassLoader classLoader = cache.get(pluginName);
+        if (classLoader == null) {
+            classLoader = new PluginClassLoader(list.toArray(new URL[list.size()]), null);
+            cache.put(pluginName, classLoader);
+        }
+        return classLoader.loadClass(className);
+    }
+
+    public static void closeURLClassLoader() {
+        for (Entry<String, PluginClassLoader> entry : cache.entrySet()) {
+            try {
+                entry.getValue().close();
+            } catch (IOException e) {
+                LOGGER.error("", e);
+            }
+        }
+    }
 }

hdata-core/src/main/java/com/github/stuxuhai/hdata/util/TypeConvertUtils.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.util;
 
 import java.math.BigDecimal;

hdata-core/src/main/java/com/github/stuxuhai/hdata/util/Utils.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月26日 下午4:35:16
- */
 package com.github.stuxuhai.hdata.util;
 
 import java.util.ArrayList;
@@ -20,75 +13,75 @@ import com.google.common.base.Throwables;
 
 public class Utils {
 
-	/**
-	 * 线程休眠
-	 *
-	 * @param millis
-	 */
-	public static void sleep(long millis) {
-		try {
-			Thread.sleep(millis);
-		} catch (InterruptedException e) {
-			Throwables.propagate(e);
-		}
-	}
+    /**
+     * 线程休眠
+     *
+     * @param millis
+     */
+    public static void sleep(long millis) {
+        try {
+            Thread.sleep(millis);
+        } catch (InterruptedException e) {
+            Throwables.propagate(e);
+        }
+    }
 
-	public static List<String> getColumns(String[] columns, String[] excludeColumns) {
-		if (excludeColumns == null || excludeColumns.length < 1) {
-			return columns == null ? null : Arrays.asList(columns);
-		}
+    public static List<String> getColumns(String[] columns, String[] excludeColumns) {
+        if (excludeColumns == null || excludeColumns.length < 1) {
+            return columns == null ? null : Arrays.asList(columns);
+        }
 
-		List<String> list = new ArrayList<String>();
-		for (String column : columns) {
-			if (!ArrayUtils.contains(excludeColumns, column)) {
-				list.add(column);
-			}
-		}
-		return list;
-	}
+        List<String> list = new ArrayList<String>();
+        for (String column : columns) {
+            if (!ArrayUtils.contains(excludeColumns, column)) {
+                list.add(column);
+            }
+        }
+        return list;
+    }
 
-	public static List<String> getColumns(List<String> columns, String[] excludeColumns) {
-		return getColumns(columns.toArray(new String[columns.size()]), excludeColumns);
-	}
+    public static List<String> getColumns(List<String> columns, String[] excludeColumns) {
+        return getColumns(columns.toArray(new String[columns.size()]), excludeColumns);
+    }
 
-	/**
-	 * 修复HDFS路径（将主机名改成IP）
-	 *
-	 * @param srcLocaltion
-	 * @param metastoreUris
-	 * @return
-	 */
-	public static String fixLocaltion(String srcLocaltion, String metastoreUris) {
-		Matcher ipMatcher = Pattern.compile("(\\d+\\.){3}\\d+").matcher(metastoreUris.split(",")[0].trim());
-		if (ipMatcher.find()) {
-			String masterIP = ipMatcher.group();
-			return srcLocaltion.replaceFirst("^hdfs://\\w+:", "hdfs://" + masterIP + ":");
-		}
-		return srcLocaltion;
-	}
+    /**
+     * 修复HDFS路径（将主机名改成IP）
+     *
+     * @param srcLocaltion
+     * @param metastoreUris
+     * @return
+     */
+    public static String fixLocaltion(String srcLocaltion, String metastoreUris) {
+        Matcher ipMatcher = Pattern.compile("(\\d+\\.){3}\\d+").matcher(metastoreUris.split(",")[0].trim());
+        if (ipMatcher.find()) {
+            String masterIP = ipMatcher.group();
+            return srcLocaltion.replaceFirst("^hdfs://\\w+:", "hdfs://" + masterIP + ":");
+        }
+        return srcLocaltion;
+    }
 
-	/**
-	 * 解析分区值
-	 *
-	 * @param partitions
-	 * @return
-	 */
-	public static List<String> parsePartitionValue(String partitions) {
-		List<String> partitionValues = new ArrayList<String>();
-		String[] partitionKeyValue = partitions.split("\\s*,\\s*");
-		for (String kv : partitionKeyValue) {
-			String[] tokens = StringUtils.splitPreserveAllTokens(kv, "=");
-			partitionValues.add(tokens[1]);
-		}
-		return partitionValues;
-	}
+    /**
+     * 解析分区值
+     *
+     * @param partitions
+     * @return
+     */
+    public static List<String> parsePartitionValue(String partitions) {
+        List<String> partitionValues = new ArrayList<String>();
+        String[] partitionKeyValue = partitions.split("\\s*,\\s*");
+        for (String kv : partitionKeyValue) {
+            String[] tokens = StringUtils.splitPreserveAllTokens(kv, "=");
+            partitionValues.add(tokens[1]);
+        }
+        return partitionValues;
+    }
 
-	/**
-	 * 获取配置目录
-	 *
-	 * @return
-	 */
-	public static String getConfigDir() {
-		return System.getProperty("hdata.conf.dir") + System.getProperty("file.separator");
-	}
+    /**
+     * 获取配置目录
+     *
+     * @return
+     */
+    public static String getConfigDir() {
+        return System.getProperty("hdata.conf.dir") + System.getProperty("file.separator");
+    }
 }

hdata-csv/src/main/java/com/github/stuxuhai/hdata/plugin/writer/csv/CSVWriter.java

@@ -1,10 +1,3 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月27日 下午4:02:30
- */
 package com.github.stuxuhai.hdata.plugin.writer.csv;
 
 import java.io.FileOutputStream;
@@ -31,132 +24,128 @@ import com.github.stuxuhai.hdata.api.Writer;
 import com.github.stuxuhai.hdata.exception.HDataException;
 import com.google.common.base.Preconditions;
 
-/**
- * @author wuya
- *
- */
 public class CSVWriter extends Writer {
 
-	private String path = null;
-	private String encoding = null;
-	private String separator = null;
-	private java.io.Writer writer;
-	private CSVPrinter csvPrinter;
-	private Fields fields;
-	private boolean showColumns;
-	private boolean showTypesAndComments;
-	private String[] types;
-	private String[] comments;
-	private List<Object> csvList = new ArrayList<Object>();
-	private static AtomicInteger sequence = new AtomicInteger(0);
-	private DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
-	private static final Pattern REG_FILE_PATH_WITHOUT_EXTENSION = Pattern.compile(".*?(?=\\.\\w+$)");
-	private static final Pattern REG_FILE_EXTENSION = Pattern.compile("(\\.\\w+)$");
-
-	@Override
-	public void prepare(JobContext context, PluginConfig writerConfig) {
-		path = writerConfig.getString(CSVWriterProperties.PATH);
-		Preconditions.checkNotNull(path, "CSV writer required property: path");
-
-		encoding = writerConfig.getString(CSVWriterProperties.ENCODING, "UTF-8");
-		separator = StringEscapeUtils.unescapeJava(writerConfig.getString(CSVWriterProperties.SEPARATOR, ","));
-
-		fields = context.getFields();
-		showColumns = writerConfig.getBoolean(CSVWriterProperties.SHOW_COLUMNS, false);
-		showTypesAndComments = writerConfig.getBoolean(CSVWriterProperties.SHOW_TYPES_AND_COMMENTS, false);
-		if (showTypesAndComments) {
-			types = context.getJobConfig().getString("types").split("\001");
-			comments = context.getJobConfig().getString("comments").split("\001");
-		}
-
-		int parallelism = writerConfig.getParallelism();
-		if (parallelism > 1) {
-			String filePathWithoutExtension = "";
-			String fileExtension = "";
-			Matcher m1 = REG_FILE_PATH_WITHOUT_EXTENSION.matcher(path.trim());
-			if (m1.find()) {
-				filePathWithoutExtension = m1.group();
-			}
-
-			Matcher m2 = REG_FILE_EXTENSION.matcher(path.trim());
-			if (m2.find()) {
-				fileExtension = m2.group();
-			}
-			path = String.format("%s_%04d%s", filePathWithoutExtension, sequence.getAndIncrement(), fileExtension);
-		}
-
-		try {
-			writer = new OutputStreamWriter(new FileOutputStream(path), encoding);
-		} catch (Exception e) {
-			throw new HDataException(e);
-		}
-	}
-
-	@Override
-	public void execute(Record record) {
-		if (csvPrinter == null) {
-			try {
-				csvPrinter = new CSVPrinter(writer, CSVFormat.DEFAULT.withDelimiter(separator.charAt(0)));
-				if (showTypesAndComments) {
-					for (String type : types) {
-						csvList.add(type);
-					}
-					csvPrinter.printRecord(csvList);
-					csvList.clear();
-
-					for (String comment : comments) {
-						csvList.add(comment);
-					}
-					csvPrinter.printRecord(csvList);
-					csvList.clear();
-				}
-
-				if (showColumns) {
-					for (Object object : fields) {
-						csvList.add(object);
-					}
-					csvPrinter.printRecord(csvList);
-					csvList.clear();
-				}
-			} catch (IOException e) {
-				throw new HDataException(e);
-			}
-		}
-
-		for (int i = 0, len = record.size(); i < len; i++) {
-			Object obj = record.get(i);
-			if (obj instanceof Timestamp) {
-				csvList.add(dateFormat.format(obj));
-			} else {
-				csvList.add(obj);
-			}
-		}
-
-		try {
-			csvPrinter.printRecord(csvList);
-		} catch (IOException e) {
-			throw new HDataException(e);
-		}
-		csvList.clear();
-	}
-
-	@Override
-	public void close() {
-		if (csvPrinter != null) {
-			try {
-				csvPrinter.close();
-			} catch (IOException e) {
-				throw new HDataException(e);
-			}
-		}
-
-		if (writer != null) {
-			try {
-				writer.close();
-			} catch (IOException e) {
-				throw new HDataException(e);
-			}
-		}
-	}
+    private String path = null;
+    private String encoding = null;
+    private String separator = null;
+    private java.io.Writer writer;
+    private CSVPrinter csvPrinter;
+    private Fields fields;
+    private boolean showColumns;
+    private boolean showTypesAndComments;
+    private String[] types;
+    private String[] comments;
+    private List<Object> csvList = new ArrayList<Object>();
+    private static AtomicInteger sequence = new AtomicInteger(0);
+    private DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+    private static final Pattern REG_FILE_PATH_WITHOUT_EXTENSION = Pattern.compile(".*?(?=\\.\\w+$)");
+    private static final Pattern REG_FILE_EXTENSION = Pattern.compile("(\\.\\w+)$");
+
+    @Override
+    public void prepare(JobContext context, PluginConfig writerConfig) {
+        path = writerConfig.getString(CSVWriterProperties.PATH);
+        Preconditions.checkNotNull(path, "CSV writer required property: path");
+
+        encoding = writerConfig.getString(CSVWriterProperties.ENCODING, "UTF-8");
+        separator = StringEscapeUtils.unescapeJava(writerConfig.getString(CSVWriterProperties.SEPARATOR, ","));
+
+        fields = context.getFields();
+        showColumns = writerConfig.getBoolean(CSVWriterProperties.SHOW_COLUMNS, false);
+        showTypesAndComments = writerConfig.getBoolean(CSVWriterProperties.SHOW_TYPES_AND_COMMENTS, false);
+        if (showTypesAndComments) {
+            types = context.getJobConfig().getString("types").split("\001");
+            comments = context.getJobConfig().getString("comments").split("\001");
+        }
+
+        int parallelism = writerConfig.getParallelism();
+        if (parallelism > 1) {
+            String filePathWithoutExtension = "";
+            String fileExtension = "";
+            Matcher m1 = REG_FILE_PATH_WITHOUT_EXTENSION.matcher(path.trim());
+            if (m1.find()) {
+                filePathWithoutExtension = m1.group();
+            }
+
+            Matcher m2 = REG_FILE_EXTENSION.matcher(path.trim());
+            if (m2.find()) {
+                fileExtension = m2.group();
+            }
+            path = String.format("%s_%04d%s", filePathWithoutExtension, sequence.getAndIncrement(), fileExtension);
+        }
+
+        try {
+            writer = new OutputStreamWriter(new FileOutputStream(path), encoding);
+        } catch (Exception e) {
+            throw new HDataException(e);
+        }
+    }
+
+    @Override
+    public void execute(Record record) {
+        if (csvPrinter == null) {
+            try {
+                csvPrinter = new CSVPrinter(writer, CSVFormat.DEFAULT.withDelimiter(separator.charAt(0)));
+                if (showTypesAndComments) {
+                    for (String type : types) {
+                        csvList.add(type);
+                    }
+                    csvPrinter.printRecord(csvList);
+                    csvList.clear();
+
+                    for (String comment : comments) {
+                        csvList.add(comment);
+                    }
+                    csvPrinter.printRecord(csvList);
+                    csvList.clear();
+                }
+
+                if (showColumns) {
+                    for (Object object : fields) {
+                        csvList.add(object);
+                    }
+                    csvPrinter.printRecord(csvList);
+                    csvList.clear();
+                }
+            } catch (IOException e) {
+                throw new HDataException(e);
+            }
+        }
+
+        for (int i = 0, len = record.size(); i < len; i++) {
+            Object obj = record.get(i);
+            if (obj instanceof Timestamp) {
+                csvList.add(dateFormat.format(obj));
+            } else {
+                csvList.add(obj);
+            }
+        }
+
+        try {
+            csvPrinter.printRecord(csvList);
+        } catch (IOException e) {
+            throw new HDataException(e);
+        }
+        csvList.clear();
+    }
+
+    @Override
+    public void close() {
+        if (csvPrinter != null) {
+            try {
+                csvPrinter.close();
+            } catch (IOException e) {
+                throw new HDataException(e);
+            }
+        }
+
+        if (writer != null) {
+            try {
+                writer.close();
+            } catch (IOException e) {
+                throw new HDataException(e);
+            }
+        }
+    }
 
 }

hdata-csv/src/main/java/com/github/stuxuhai/hdata/plugin/writer/csv/CSVWriterProperties.java

@@ -1,20 +1,9 @@
-/*
- * 蘑菇街 Inc.
- * Copyright (c) 2010-2014 All Rights Reserved.
- *
- * Author: wuya
- * Create Date: 2014年6月27日 下午4:03:58
- */
 package com.github.stuxuhai.hdata.plugin.writer.csv;
 
-/**
- * @author wuya
- *
- */
 public class CSVWriterProperties {
-	public static final String PATH = "path";
-	public static final String ENCODING = "encoding";
-	public static final String SEPARATOR = "separator";
-	public static final String SHOW_COLUMNS = "show.columns";
-	public static final String SHOW_TYPES_AND_COMMENTS = "show.types.and.comments";
+    public static final String PATH = "path";
+    public static final String ENCODING = "encoding";
+    public static final String SEPARATOR = "separator";
+    public static final String SHOW_COLUMNS = "show.columns";
+    public static final String SHOW_TYPES_AND_COMMENTS = "show.types.and.comments";
 }