189 8069 5689
这篇文章将为大家详细讲解有关Storm如何和Kafka进行整合，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。
创新互联专注于企业成都全网营销推广、网站重做改版、义县网站定制设计、自适应品牌网站建设、H5场景定制、商城网站开发、集团公司官网建设、成都外贸网站建设、高端网站制作、响应式网页设计等建站业务，价格优惠性价比高，为义县等各大城市提供网站开发制作服务。
对于Storm 如何和Kafka进行整合
package com.mixbox.storm.kafka;
import backtype.storm.Config;
import backtype.storm.metric.api.IMetric;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import kafka.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.mixbox.storm.kafka.PartitionManager.KafkaMessageId;
import java.util.*;
/**
* @author Yin Shuai
*/
public class KafkaSpout extends BaseRichSpout {
public static final Logger LOG = LoggerFactory.getLogger(KafkaSpout.class);
/**
* 内部类，Message和Offset的偏移量对象
*
* @author Yin Shuai
*/
public static class MessageAndRealOffset {
public Message msg;
public long offset;
public MessageAndRealOffset(Message msg, long offset) {
this.msg = msg;
this.offset = offset;
}
}
/**
* 发射的枚举类
* @author Yin Shuai
*/
static enum EmitState {
EMITTED_MORE_LEFT, EMITTED_END, NO_EMITTED
}
String _uuid = UUID.randomUUID().toString();
SpoutConfig _spoutConfig;
SpoutOutputCollector _collector;
// 分区的协调器，getMyManagedPartitions 拿到我所管理的分区
PartitionCoordinator _coordinator;
// 动态的分区链接：保存到kafka各个节点的连接，以及负责的topic的partition号码
DynamicPartitionConnections _connections;
// 提供了从zookeeper读写kafka 消费者信息的功能
ZkState _state;
// 上次更新的毫秒数
long _lastUpdateMs = 0;
// 当前的分区
int _currPartitionIndex = 0;
public KafkaSpout(SpoutConfig spoutConf) {
_spoutConfig = spoutConf;
}
@SuppressWarnings("unchecked")
@Override
public void open(Map conf, final TopologyContext context,
final SpoutOutputCollector collector) {
_collector = collector;
List zkServers = _spoutConfig.zkServers;
// 初始化的时候如果zkServers 为空，那么初始化 默认的配置Zookeeper
if (zkServers == null) {
zkServers = new ArrayList() {
{
add("192.168.50.144");
add("192.168.50.169");
add("192.168.50.207");
}
};
// zkServers =
// (List)conf.get(Config.STORM_ZOOKEEPER_SERVERS);
System.out.println(" 使用的是Storm默认配置的Zookeeper List ： " + zkServers);
}
Integer zkPort = _spoutConfig.zkPort;
// 在这里我们也同时 来检查zookeeper的端口是否为空
if (zkPort == null) {
zkPort = 2181;
// zkPort = ((Number)
// conf.get(Config.STORM_ZOOKEEPER_PORT)).intValue();
}
Map stateConf = new HashMap(conf);
stateConf.put(Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, zkServers);
stateConf.put(Config.TRANSACTIONAL_ZOOKEEPER_PORT, zkPort);
stateConf.put(Config.TRANSACTIONAL_ZOOKEEPER_ROOT, _spoutConfig.zkRoot);
// 通过保存的配置文件，我们持有了一个zookeeper的state，支持节点内容的创建和删除
_state = new ZkState(stateConf);
// 对于连接的维护
_connections = new DynamicPartitionConnections(_spoutConfig,
KafkaUtils.makeBrokerReader(conf, _spoutConfig));
// using TransactionalState like this is a hack
// 拿到总共的任务次数
int totalTasks = context
.getComponentTasks(context.getThisComponentId()).size();
// 判断当前的主机是否是静态的statichost
if (_spoutConfig.hosts instanceof StaticHosts) {
_coordinator = new StaticCoordinator(_connections, conf,
_spoutConfig, _state, context.getThisTaskIndex(),
totalTasks, _uuid);
// 当你拿到的spoutConfig是zkhost的时候
} else {
_coordinator = new ZkCoordinator(_connections, conf, _spoutConfig,
_state, context.getThisTaskIndex(), totalTasks, _uuid);
}
context.registerMetric("kafkaOffset", new IMetric() {
KafkaUtils.KafkaOffsetMetric _kafkaOffsetMetric = new KafkaUtils.KafkaOffsetMetric(
_spoutConfig.topic, _connections);
@Override
public Object getValueAndReset() {
List pms = _coordinator
.getMyManagedPartitions();
Set latestPartitions = new HashSet();
for (PartitionManager pm : pms) {
latestPartitions.add(pm.getPartition());
}
_kafkaOffsetMetric.refreshPartitions(latestPartitions);
for (PartitionManager pm : pms) {
_kafkaOffsetMetric.setLatestEmittedOffset(
pm.getPartition(), pm.lastCompletedOffset());
}
return _kafkaOffsetMetric.getValueAndReset();
}
}, _spoutConfig.metricsTimeBucketSizeInSecs);
context.registerMetric("kafkaPartition", new IMetric() {
@Override
public Object getValueAndReset() {
List pms = _coordinator
.getMyManagedPartitions();
Map concatMetricsDataMaps = new HashMap();
for (PartitionManager pm : pms) {
concatMetricsDataMaps.putAll(pm.getMetricsDataMap());
}
return concatMetricsDataMaps;
}
}, _spoutConfig.metricsTimeBucketSizeInSecs);
}
@Override
public void close() {
_state.close();
}
@Override
public void nextTuple() {
// Storm-spout 是从kafka 消费数据,把 kafka 的 consumer
// 当成是一个spout，并且向其他的bolt的发送数据
// 拿到当前我管理的这些PartitionsManager
List managers = _coordinator.getMyManagedPartitions();
for (int i = 0; i < 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() {
// 停止工作
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();
}
}
} 在粗浅的代码阅读之后，在这里进行详细的分析：
1 KafkaSpout之中持有了一个 MessageAndRealOffset 的内部类
public static class MessageAndRealOffset
{
public Message msg;
public long offset;
public MessageAndRealOffset(Message msg,long offset)
{
this.msg = msg;
this.offset = offset;
}
}2 在Spout之中我们还持有了一个PartitionCoordinator的分区协调器，默认的情况我们实例化的对象
是ZKCoordinator
关于Storm如何和Kafka进行整合就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。