Fig

Fig（无花果）是一个根据多任务流水线模型开发的运行框架，框架多任务并发使用的java的线程池进行控制，使用队列实现任务间的数据传递

预览版 仓库 地址：https://oss.sonatype.org/content/repositories/snapshots
   maven 依赖 ：

<dependency>  <groupId>com.github.taomus.fig</groupId>  <artifactId>fig-core</artifactId>  <version>0.1.0-SNAPSHOT</version>  <type>module</type></dependency><dependency>  <groupId>com.github.taomus.fig</groupId>  <artifactId>fig-spring-plugin</artifactId>  <version>0.1.0-SNAPSHOT</version>  <type>module</type></dependency>

 Xtend 代码实例:           

package com.github.test1.stockimport com.github.taomus.fig.core.engine.Dataimport com.github.taomus.fig.core.engine.Figimport com.github.taomus.fig.core.engine.FigEngineimport com.github.taomus.fig.core.engine.Taskimport com.github.test.stock.entity.StockDataimport java.nio.charset.Charsetimport java.util.Arraysimport java.util.Vectorimport joinery.DataFrameimport org.apache.commons.lang3.SerializationUtilsimport org.jsoup.Connectionimport org.jsoup.Jsoupimport org.slf4j.LoggerFactoryimport org.springframework.util.StreamUtilsclass CollectStock1 extends Fig {	val static LOG = LoggerFactory.getLogger(CollectStock1)		def static void main(String[] args){		FigEngine.instance.addModule(CollectStock1)		FigEngine.instance.run()		FigEngine.instance.taskQueue.put(Data.create("start",null))	}		def String getUrl(String code,String date) {		if (code.startsWith("0")) {			return ''' http://quotes.money.163.com/service/chddata.html?code=1«code»&start=«date»0101&end=«date»1231			'''		} else {			return ''' http://quotes.money.163.com/service/chddata.html?code=0«code»&start=«date»0101&end=«date»1231			'''		}	}	@Task("start")	def Data[] getStockCodes(Data value){		return #[			Data.create("buildData",new StockData("平安银行","000001")),			Data.create("buildData",new StockData("万科A","000002"))		]	}		@Task("buildData")	def Data[] collect(Data value){		var stockinfo = value.getData as StockData		var results = new Vector<Data>();		for(date:2000..2020){			var info = SerializationUtils.<StockData>clone(stockinfo)			info.date = String.valueOf(date)			results.add(Data.create("stock_history",info))		}		return results	}	@Task("stock_history")	def Data startHistory(Data value) {		var d = value.getData() as StockData		var String url = getUrl(d.code,d.date)		LOG.info(url)		var Connection connection = Jsoup.connect(url);		var Connection.Response response = connection.method(Connection.Method.GET).ignoreContentType(true).timeout(10 *			1000).execute();		var a = StreamUtils.copyToString(response.bodyStream, Charset.forName("UTF-8"))		LOG.info(a)		var data = a.split("\n")		if(data.size == 1){			return null		}		var DataFrame<Object> df = new DataFrame("日期", "股票代码", "名称", "收盘价", "最高价", "最低价", "开盘价", "前收盘", "涨跌额", "涨跌幅",			"换手率", "成交量", "成交金额", "总市值", "流通市值", "成交笔数");		for(index : 1..data.length-1){			df.append(Arrays.asList(data.get(index).split(',')))		}		var results = new Vector();		var indexs = df.index();		for (Object index : indexs) {			var sd = new StockData()			sd.date = df.get(index,"日期") as String			sd.code = df.get(index,"股票代码") as String			sd.code = sd.code.substring(1)			sd.name = d.name			sd.opening = Float.valueOf(df.get(index,"开盘价") as String)			sd.ending = Float.valueOf(df.get(index,"收盘价") as String)			sd.low = Float.valueOf(df.get(index,"最低价") as String)			sd.hig = Float.valueOf(df.get(index,"最高价") as String)			results.add(sd)		}		var resdata = Data.create("calc",results.reverse);		resdata.priority = 2		return resdata	}		@Task("calc")	def Data calc(Data value) {		return null;	}	}