IBatchSpout API怎么使用

这篇文章主要介绍"IBatchSpout API怎么使用"，在日常操作中，相信很多人在IBatchSpout API怎么使用问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答"IBatchSpout API怎么使用"的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

IBatchSpout是storm trident推出的一种可以批量发射的Spout。非事务性，基本的spout

1：Map getComponentConfiguration();定义配置，可以用backtype.storm.Config。

2：void open(Map conf, TopologyContext context); Spout的初始化方法 ，参数conf即是getComponentConfiguration定义的配置

3：Fields getOutputFields(); 声明输出的fields

4：void emitBatch(long batchId, TridentCollector collector); 批量发射tuple，本次的批次号为batchId

5：void ack(long batchId);批次号为batchId的数据处理成功

6: void close();

一个例子

package storm.projectA;import java.io.BufferedReader;import java.io.FileNotFoundException;import java.io.FileReader;import java.io.IOException;import java.util.ArrayList;import java.util.HashMap;import java.util.List;import java.util.Map;import storm.trident.operation.TridentCollector;import storm.trident.spout.IBatchSpout;import backtype.storm.Config;import backtype.storm.task.TopologyContext;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Values;public class MySpout implements IBatchSpout{ /**  *   */ private static final long serialVersionUID = 1L; private long maxBatchSize;//每批次最大的数量 private BufferedReader br;//源文件流 HashMap>> batches = new HashMap>>();//保存发送过的所有数据，以便于重复发送 /**  * @param conf 配置  * @param context   */ @Override public void open(Map conf, TopologyContext context) {  String filePath = (String)conf.get("filePath");  maxBatchSize = (Long)conf.get("maxBatchSize");  try {   br = new BufferedReader(new FileReader(filePath));  } catch (FileNotFoundException e) {   e.printStackTrace();  } } /*** spout的发送方法  * @param batchId 批次id  * @param collector 批量发射器  */ @Override public void emitBatch(long batchId, TridentCollector collector) {  List> batch = batches.get(batchId);  if (batch == null) {   batch = new ArrayList>();   for (int i = 0; i < maxBatchSize; i++) {    try {     String line = br.readLine();     if(line == null){      break;     }     batch.add(new Values(line));    } catch (IOException e) {     e.printStackTrace();    }   }  }  for(List