storm中KafkaSpout的选择

---

Storm最常用的消息源就是Kafka，在对接的时候大多需要使用KafkaSpout；

在网上大概有两种KafkaSpout，一种是只有几十行，一种却有一大啪啦类文件。

---

在kafka中，同一个partition中的消息只能被同一个组的一个consumer消费，不能并发，所以kafka的并发说的是多partition的并发；

kafka的consumer API分为high level consumer和low level consumer，官方建议使用前者，以为不用关心partition、offset那些，但是后者也有其存在的意义：1.多次读取的时候；2.选择性读取部分消息；3.控制消费过程。

---

写法比较简单的KafkaSpout：

public class KafkaSpouttest implements IRichSpout {

    private static final long serialVersionUID = 1L;
    private SpoutOutputCollector collector;
    private ConsumerConnector consumer;
    private String topic;

    public KafkaSpouttest() {}

    public KafkaSpouttest(String topic) {
        this.topic = topic;
    }

    public void ack(Object arg0) {

}

    private static ConsumerConfig createConsumerConfig() {
        Properties props = new Properties();
        // 设置zookeeper的链接地址
        props.put("zookeeper.connect", "localhost:2181");
        // 设置group id
        props.put("group.id", "1");
        // kafka的group 消费记录是保存在zookeeper上的, 但这个信息在zookeeper上不是实时更新的, 需要有个间隔时间更新
        props.put("auto.commit.interval.ms", "1000");
        props.put("zookeeper.session.timeout.ms", "10000");
        return new ConsumerConfig(props);
    }

    public void activate() {
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(createConsumerConfig());
        Map < String,
        Integer > topickMap = new HashMap < String,
        Integer > ();
        topickMap.put(topic, 1);

        System.out.println("*********Results********topic:" + topic);

        Map < String,
        List < KafkaStream < byte[],
        byte[] >>> streamMap = consumer.createMessageStreams(topickMap);
        KafkaStream < byte[],
        byte[] > stream = streamMap.get(topic).get(0);
        ConsumerIterator < byte[],
        byte[] > it = stream.iterator();
        while (it.hasNext()) {
            String value = new String(it.next().message());
            SimpleDateFormat formatter = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss SSS");
            Date curDate = new Date(System.currentTimeMillis()); //获取当前时间      
            String str = formatter.format(curDate);

            System.out.println("storm接收到来自kafka的消息------->" + value);

            collector.emit(new Values(value, 1, str), value);
        }
    }

    public void close() {
        // TODO Auto-generated method stub
    }

    public void deactivate() {
        // TODO Auto-generated method stub
    }

    public void fail(Object arg0) {
        // TODO Auto-generated method stub
    }

    public void nextTuple() {
        // TODO Auto-generated method stub
    }

    public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word", "id", "time"));
    }

    public Map < String,
    Object > getComponentConfiguration() {
        System.out.println("getComponentConfiguration被调用");
        topic = "admln";
        return null;
    }

}

方法相关的不解释，和本主题相关的一句话是：

byte[] >>> streamMap = consumer.createMessageStreams(topickMap);

想说的是它用的是High Level API

---

复杂的代码就多了，在github上有好几个，最官方的还是apache storm自带的：

里面和本主题相关的一句话是DynamicPartitionConnections.java中的60行：

_connections.put(host, new ConnectionInfo(new SimpleConsumer(host.host, host.port, _config.socketTimeoutMs, _config.bufferSizeBytes, _config.clientId)));

它用的是low level API

---

apache KafkaSpout 在 topology 中的配置

String zkConnString = "node1:2181,node2:2181,node3:2181";
        String topicName = "testtopic";
        BrokerHosts hosts = new ZkHosts(zkConnString);
        SpoutConfig spoutConfig = new SpoutConfig(hosts, topicName, "/" + topicName, UUID.randomUUID().toString());
        spoutConfig.forceFromStart = false;
        spoutConfig.zkPort = 2181;
        spoutConfig.zkServers = Arrays.asList(new String[]{"node1","node2","node3"});
        
        spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
        
        KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
        
        TopologyBuilder builder = new TopologyBuilder();
        // 构造NC数据流向图
        builder.setSpout("mrspout", kafkaSpout, 30);
        builder.setBolt("mrverifybolt", new MRVerifyBolt(), 30)
                .shuffleGrouping("mrspout");
        builder.setBolt("mr2storagebolt", new MR2StorageBolt(), 30)
                .shuffleGrouping("mrverifybolt");
        // 以类名作为STORM任务名
        String name = MRTopology.class.getSimpleName();
        // 传主机名则为集群运行模式，不传则为本地运行模式
        if (args != null && args.length > 0) {
            Config conf = new Config();
            // 通过指定nimbus主机
            conf.put(Config.NIMBUS_HOST, args[0]);
            conf.setNumWorkers(6);
            conf.setNumAckers(0);
            conf.setMaxSpoutPending(100000);
            StormSubmitter.submitTopologyWithProgressBar(name, conf,
                    builder.createTopology());
        } else {
            Map conf = new HashMap();
            conf.put(Config.TOPOLOGY_WORKERS, 1);
            conf.put(Config.TOPOLOGY_DEBUG, true);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(name, conf, builder.createTopology());
        }
    }

关于 spoutConfig.servers 和 spoutConfig.port 在实际应用中其实不设置也可以，因为在集群中如果不设置 storm 默认会把 storm 配置中的 zookeeper 地址和端口，设置的用处是在 eclipse 中测试运行的时候因为是模拟 storm cluster， 所以主动设置。

---

两者各有优劣，相同点性能，简单测试过，low level的要好点，但是相差不大（都在合适的配置下，小集群）；

不同点是high level 的代码简单，而low level的代码很多，配置也多，用着麻烦(也不是很麻烦)；

low level的优点是支持重读，就是配置中的 spoutConfig.forceFromStart = false; ，支持重读的另一个好处是和storm的acker结合，可以重发，防止丢数据，这一点比low level的要安全一点，另一个好处是配置多，使用就很难灵活，比如设置KafkaSpout的fetchSizeBytes，和kafka的bufferSizeBytes对应，是优化的一个手段。

至于选择哪种，支持后者，反正storm中已经自带了，不需要自己写，配置就好，而且0.9.4中优化了很多KafkaSpout的问题。

---