kafka参数配置详情：
public final BrokerHosts hosts;//设置kafka从哪里获取相关的配置信息 public final String topic;//从哪个topic开始消费 public final String clientId;//设置客户端标识 public int fetchSizeBytes = 1024 * 1024;//发给Kafka的每个FetchRequest中，用此指定想要的response中总的消息的大小 public int socketTimeoutMs = 10000;//设置的超时时间 public int fetchMaxWait = 10000;//设置的在broker无消息时的等待时间 public int bufferSizeBytes = 1024 * 1024;//SimpleConsumer所使用的SocketChannel的读缓冲区大小 public MultiScheme scheme = new RawMultiScheme();//设置从服务器读取的byte[]流反序列化方式 public boolean ignoreZkOffsets = false;//是否强制从Kafka中offset最小的开始读起 public long startOffsetTime = kafka.api.OffsetRequest.EarliestTime();//从哪里的offset开始读取消息，默认从消息的最前端开始，有两种方式可选 public long maxOffsetBehind = Long.MAX_VALUE;//KafkaSpout读取的进度与目标进度相差多少，相差太多，Spout会丢弃中间的消息 public boolean useStartOffsetTimeIfOffsetOutOfRange = true;//如果所请求的offset对应的消息在Kafka中不存在，是否使用startOffsetTime public int metricsTimeBucketSizeInSecs = 60;//多长时间统计一次消息
代码示例：注意事项（kafka是scala编写，所有依赖scala环境，一定要统一scala版本，本次使用的为scala2.10.1）
package kafka; import java.util.Arrays; import java.util.HashMap; import java.util.Iterator; import java.util.Map; import java.util.Map.Entry; import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.storm.Config; import org.apache.storm.LocalCluster; import org.apache.storm.StormSubmitter; import org.apache.storm.generated.AlreadyAliveException; import org.apache.storm.generated.AuthorizationException; import org.apache.storm.generated.InvalidTopologyException; import org.apache.storm.kafka.*; import org.apache.storm.spout.SchemeAsMultiScheme; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; import org.apache.storm.topology.TopologyBuilder; import org.apache.storm.topology.base.BaseRichBolt; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Tuple; import org.apache.storm.tuple.Values;
/** * Created by shea on 2018/2/2. */ public class KafkaTopology2 {
1public static class KafkaWordSplitter extends BaseRichBolt { 2 3 private static final Log LOG = LogFactory.getLog(KafkaWordSplitter.class); 4 private static final long serialVersionUID = 886149197481637894L; 5 private OutputCollector collector; 6 7 public void prepare(Map stormConf, TopologyContext context, 8 OutputCollector collector) { 9 this.collector = collector; 10 } 11 12 public void execute(Tuple input) { 13 String line = input.getString(0); 14 LOG.info("RECV\[kafka -> splitter\] " + line); 15 String\[\] words = line.split("\\\\s+"); 16 for(String word : words) { 17 LOG.info("EMIT\[splitter -> counter\] " + word); 18 collector.emit(input, new Values(word, 1)); 19 } 20 collector.ack(input); 21 } 22 23 public void declareOutputFields(OutputFieldsDeclarer declarer) { 24 declarer.declare(new Fields("word", "count")); 25 } 26 27} 28 29public static class WordCounter extends BaseRichBolt { 30 31 private static final Log LOG = LogFactory.getLog(WordCounter.class); 32 private static final long serialVersionUID = 886149197481637894L; 33 private OutputCollector collector; 34 private Map<String, AtomicInteger> counterMap; 35 36 public void prepare(Map stormConf, TopologyContext context, 37 OutputCollector collector) { 38 this.collector = collector; 39 this.counterMap = new HashMap<String, AtomicInteger>(); 40 } 41 42 public void execute(Tuple input) { 43 String word = input.getString(0); 44 int count = input.getInteger(1); 45 LOG.info("RECV\[splitter -> counter\] " + word + " : " + count); 46 AtomicInteger ai = this.counterMap.get(word); 47 if(ai ==null ) { 48 ai = new AtomicInteger(); 49 this.counterMap.put(word, ai); 50 } 51 ai.addAndGet(count); 52 collector.ack(input); 53 LOG.info("CHECK statistics map: " + this.counterMap); 54 } 55 56 @Override 57 public void cleanup() { 58 LOG.info("The final result:"); 59 Iterator<Entry<String, AtomicInteger>> iter = this.counterMap.entrySet().iterator(); 60 while(iter.hasNext()) { 61 Entry<String, AtomicInteger> entry = iter.next(); 62 LOG.info(entry.getKey() + "\\t:\\t" + entry.getValue().get()); 63 } 64 65 } 66 67 public void declareOutputFields(OutputFieldsDeclarer declarer) { 68 declarer.declare(new Fields("word", "count")); 69 } 70} 71 72public static void main(String\[\] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException,AuthorizationException { 73 String zks = "data1:2181,data2:2181,data3:2181"; 74 //String topic = "my-replicated-topic5"; 75 String topic = "test"; 76 String zkRoot = "/storm"; // default zookeeper root configuration for storm 77 String id = "word"; 78 79 BrokerHosts brokerHosts = new ZkHosts(zks); 80 SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, zkRoot, id); 81 spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme()); 82 //spoutConf.forceFromStart = false;//该配置是指，如果该Topology因故障停止处理，下次正常运行时是否从Spout对应数据源Kafka//中的该订阅Topic的起始位置开始读取，如果forceFromStart=true，则之前处理过的Tuple还要重新处理一遍，否则会从上次处理的位置//继续处理，保证Kafka中的Topic数据不被重复处理，是在数据源的位置进行状态记录 83 spoutConf.zkServers = Arrays.asList(new String\[\] {"data1", "data2", "data3"}); 84 spoutConf.zkPort = 2181; 85 86 TopologyBuilder builder = new TopologyBuilder(); 87 builder.setSpout("kafka-reader", new KafkaSpout(spoutConf), 5); // Kafka我们创建了一个5分区的Topic，这里并行度设置为5 88 builder.setBolt("word-splitter", new KafkaWordSplitter(), 2).shuffleGrouping("kafka-reader"); 89 builder.setBolt("word-counter", new WordCounter()).fieldsGrouping("word-splitter", new Fields("word")); 90 91 Config conf = new Config(); 92 93 String name = MyKafkaTopology.class.getSimpleName(); 94 if (args !=null && args.length > 0) { 95 // Nimbus host name passed from command line 96 conf.put(Config.NIMBUS\_HOST, args\[0\]); 97 conf.setNumWorkers(3); 98 StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology()); 99 } else { 100 conf.setMaxTaskParallelism(3); 101 LocalCluster cluster = new LocalCluster(); 102 cluster.submitTopology(name, conf, builder.createTopology()); 103 Thread.sleep(60000); 104 cluster.shutdown(); 105 } 106}
}
/** * 获取数据，处理数据，发送数据 * ack机制即， spout发送的每一条消息，
在规定的时间内，spout收到Acker的ack响应，即认为该tuple 被后续bolt成功处理 在规定的时间内，没有收到Acker的ack响应tuple，就触发fail动作，即认为该tuple处理失败， 或者收到Acker发送的fail响应tuple，也认为失败，触发fail动作
另外Ack机制还常用于限流作用： 为了避免spout发送数据太快，而bolt处理太慢，常常设置pending数， 当spout有等于或超过pending数的tuple没有收到ack或fail响应时，跳过执行nextTuple， 从而限制spout发送数据。
网络上另外对kafka配置的解释-----摘自https://www.cnblogs.com/devos/p/4335302.html
public final BrokerHosts hosts; //用以获取Kafka broker和partition的信息 public final String topic;//从哪个topic读取消息 public final String clientId; // SimpleConsumer所用的client id
1public int fetchSizeBytes = 1024 \* 1024; //发给Kafka的每个FetchRequest中，用此指定想要的response中总的消息的大小 2public int socketTimeoutMs = 10000;//与Kafka broker的连接的socket超时时间 3public int fetchMaxWait = 10000; //当服务器没有新消息时，消费者会等待这些时间 4public int bufferSizeBytes = 1024 \* 1024;//SimpleConsumer所使用的SocketChannel的读缓冲区大小 5public MultiScheme scheme = new RawMultiScheme();//从Kafka中取出的byte\[\]，该如何反序列化 6public boolean forceFromStart = false;//是否强制从Kafka中offset最小的开始读起 7public long startOffsetTime = kafka.api.OffsetRequest.EarliestTime();//从何时的offset时间开始读，默认为最旧的offset 8public long maxOffsetBehind = Long.MAX\_VALUE;//KafkaSpout读取的进度与目标进度相差多少，相差太多，Spout会丢弃中间的消息
public boolean useStartOffsetTimeIfOffsetOutOfRange = true;//如果所请求的offset对应的消息在Kafka中不存在，是否使用startOffsetTime public int metricsTimeBucketSizeInSecs = 60;//多长时间统计一次metrics
对Zookeeper的使用
KafkaSpout的配置中有两个地方可以用到Zookeeper
- 用Zookeeper来记录KafkaSpout的处理进度，在topology重新提交或者task重启后继续之前的处理进度。在SpoutConfig中的zkServers, zkPort和zkRoot与此相关。如果zkServer和zkPort没有设置，那么KafkaSpout会使用Storm集群所用的Zookeeper记录这些信息。
- 用Zookeeper来获取Kafka中一个topic的所有partition，和每个partition的leader。这需要实现BrokerHosts的子类ZkHosts.但是，这个Zookeepr是可选的。如果使用BrokerHosts的另一个子类StaticHosts,把partition和leader的对应关系硬编码，则不需要Zookeeper来提供此功能。KafkaSpout会从Kafka集群使用的Zookeeper中提取partition和leader的对应关系。而且：
- 如果使用StatisHosts，那么KafkaSpout会使用StaticCoordinator，这个coordinator不能响应partition leader的变化。
- 如果使用ZkHosts，那么KafkaSpout会使用ZkCoordinator, 当其refresh()方法被调用后，这个cooridnator会检查发生leader变更的partition，并为之生成新的PartitionManager.从而能够在leader变更后，继续读取消息。
影响初始读取进度的配置项
在一个topology上线后，它从哪个offset开始读取消息呢？有一些配置项对此有影响：
- SpoutConfig中的id字段。如果想要一个topology从另一个topology之前的处理进度继续处理，它们需要有相同的id。
- KafkaConfig的forceFromStart字段。如果此字段设为true, 那么它一个topology上线后，它会忽略之前相同id的topology的进度，并且从Kafka中最早的消息开始处理。
- KafkaConfig的startOffsetTime字段。默认为kafka.api.OffsetRequest.EarliestTime()开始读，也就是从Kafka中最早的消息开始处理。也可以设成kafka.api.OffsetRequest.LatestOffset,也就是最早的消息开始读。也可以自己指定具体的值。
- KafkaConfig的maxOffsetBehind字段。这个字段对于KafkaSpout的多个处理流程都有影响。当提交一个新topology时，如果没有forceFromStart, 当KafkaSpout对某个partition的处理进度落后startOffsetTime对应的offset多于此值时，KafkaSpout会丢弃中间的消息，从而强制赶上目标进度.比如，如果startOffsetTime设成了lastestTime，那么如果进度落后超过maxOffsetBehind，KafkaSpout会直接从latestTime对应的offset开始处理。如果设成了froceFromStart，则在提交新任务时，始终会从EarliestTime开始读。
- KafkaSpout的userStartOffsetTimeIfOffsetOutOfRange字段。如果设成true，那么当fetch消息时出错，且FetchResponse显示的出错原因是OFFSET_OUT_OF_RANGE，那么就会尝试从KafkaSpout指定的startOffsetTime对应的消息开始读。例如，如果有一批消息因为超过了保存期限被Kafka删除，并且zk里记录的消息在这批被删除的消息里。如果KafkaSpout试图从zk的记录继续读，那么就会出现OFFSET_OUT_OF_RANGE的错误，从而触发这个配置。
实际上maxOffsetBehind有时候有点名不符实。当startOffsetTime为A, zk里的进度为B， A - B > maxOffsetBehind时，应该从A - maxOffsetBehind除开始读或许更好一些，而不是直接跳到startOffsetTime。此处的逻辑参见PartitionManager的实现。