业务场景：

storm+kafka 作为经典组合，storm的作用主要是进行流式计算，对于源源不断的均匀数据流流入处理是非常有效的，而现实生活中大部分场景并不是均匀的数据流，而是时而多时而少的数据流入，这种情况下显然用批量处理是不合适的，如果使用storm做实时计算的话可能因为数据拥堵而导致服务器挂掉，应对这种情况，使用kafka作为消息队列是非常合适的选择，kafka可以将不均匀的数据转换成均匀的消息流，从而和storm比较完善的结合，这样才可以实现稳定的流式计算。

废话不多说，直接low一个小demo：

Topology：

1. public class KafkaTopology {
3. private String topic ;
4. private String zkRoot;
5. private String spoutId;
7. public KafkaTopology() {}
9. public KafkaTopology(String topic,String zkRoot, String spoutId){
10. this.topic = topic;
11. this.zkRoot = zkRoot;
12. this.spoutId = spoutId;
13. }
15. public void run(){
16. //指定zk的地址
17. BrokerHosts brokerHosts = new ZkHosts(“storm1:2181,storm2:2181,storm3:2181”);
18. //配置spout
19. SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, topic, zkRoot, spoutId);
20. spoutConfig.forceFromStart = true; //从什么位置读取
21. spoutConfig.scheme = new SchemeAsMultiScheme(new MessageScheme());
23. TopologyBuilder builder = new TopologyBuilder();
24. ////设置一个spout用来从kaflka消息队列中读取数据并发送给下一级的bolt组件，
25. //此处用的spout组件并非自定义的，而是storm中已经开发好的KafkaSpout
26. builder.setSpout(spoutId, new KafkaSpout(spoutConfig));
27. builder.setBolt(“wordSplit” , new WordSplit()).shuffleGrouping(spoutId);
28. builder.setBolt(“writer”, new WriterBolt(),4).fieldsGrouping(“wordSplit”, new Fields(“word”));
30. Config conf = new Config();
31. conf.setNumWorkers(4);
32. conf.setNumAckers(0);
33. conf.setDebug(false);
35. //LocalCluster用来将topology提交到本地模拟器运行，方便开发调试
36. LocalCluster localCluster = new LocalCluster();
37. localCluster.submitTopology(“WordCount”, conf, builder.createTopology());
40. //提交topology到storm集群中运行
41. // StormSubmitter.submitTopology(“wordCount”, conf, builder.createTopology());
42. }
44. public static void main(String[] args) {
45. new KafkaTopology(“test”, “/kafka-storm”, “KafkaSpout”).run();
46. }
47. }

WordSplit:

1. public class WordSplit extends BaseBasicBolt {
4. private static final long serialVersionUID = 1L;
6. @Override
7. public void execute(Tuple tuple, BasicOutputCollector collector) {
8. String line = tuple.getString(0);
9. String[] words = line.split(” “);
10. for (String word : words) {
11. //判断某字符串是否不为空且长度不为0且不由空白符(whitespace)构成
12. if(StringUtils.isNotBlank(word.trim())){
13. word = word.toUpperCase();
14. collector.emit(new Values(word));
15. }
16. }
17. }
19. @Override
20. public void declareOutputFields(OutputFieldsDeclarer declarer) {
21. declarer.declare(new Fields(“word”));
22. }
24. }

WirterBolt:

1. // 将数据写入文件
2. public class WriterBolt extends BaseBasicBolt{
4. private static final long serialVersionUID = 1L;
5. FileWriter writer = null;
7. @Override
8. public void prepare(Map conf, TopologyContext context) {
9. try {
10. writer = new FileWriter(“e:/”+“wordcount-“ +UUID.randomUUID().toString());
11. } catch (IOException e) {
12. e.printStackTrace();
13. }
14. }
16. @Override
17. public void execute(Tuple tuple, BasicOutputCollector collector) {
18. String line = tuple.getString(0);
19. try {
20. writer.write(line);
21. writer.write(“\n”);
22. writer.flush();
23. } catch (IOException e) {
24. e.printStackTrace();
25. }
26. }
28. @Override
29. public void declareOutputFields(OutputFieldsDeclarer declarer) {
31. }
33. }

本demo主要是standlocal模式，生产环境一些参数的优化不作解释·~~~