Shulou(Shulou.com)05/31 Report--

< managers.size(); i++) { // 对于每一个分区的 PartitionManager // in case the number of managers decreased // 当前的分区 _currPartitionIndex = _currPartitionIndex % managers.size(); // 拿到当前的分区，并且发送，这里把SpoutOutputCollector传递进去了，由他发射元祖 EmitState state = managers.get(_currPartitionIndex) .next(_collector); // 如果发送状态为：发送-还有剩余 if (state != EmitState.EMITTED_MORE_LEFT) { _currPartitionIndex = (_currPartitionIndex + 1) % managers.size(); } // 如果发送的状态为： 发送-没有剩余 if (state != EmitState.NO_EMITTED) { break; } } long now = System.currentTimeMillis(); if ((now - _lastUpdateMs) >

_ spoutConfig.stateUpdateIntervalMs) {commit ();}} @ Override public void ack (Object msgId) {KafkaMessageId id = (KafkaMessageId) msgId; PartitionManager m = _ coordinator.getManager (id.partition); if (m! = null) {m.ack (id.offset) } @ Override public void fail (Object msgId) {KafkaMessageId id = (KafkaMessageId) msgId; PartitionManager m = _ coordinator.getManager (id.partition); if (m! = null) {m.fail (id.offset) } @ Override public void deactivate () {/ / stop working commit ();} @ Override public void declareOutputFields (OutputFieldsDeclarer declarer) {System.out.println (_ spoutConfig.scheme.getOutputFields ()); declarer.declare (_ spoutConfig.scheme.getOutputFields ()) } private void commit () {_ lastUpdateMs = System.currentTimeMillis (); for (PartitionManager manager: _ coordinator.getMyManagedPartitions ()) {manager.commit ();}

After reading the shallow code, analyze it in detail here:

1 KafkaSpout holds an inner class of MessageAndRealOffset

Public static class MessageAndRealOffset {public Message msg; public long offset; public MessageAndRealOffset (Message msg,long offset) {this.msg = msg; this.offset = offset;}}

2 in Spout we also hold a PartitionCoordinator partition coordinator, which we instantiate by default

It's ZKCoordinator.

On how to integrate Storm and Kafka to share here, I hope that the above content can be of some help to you, can learn more knowledge. If you think the article is good, you can share it for more people to see.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.