13518219792
小编给大家分享一下Storm-kafka中如何实现一个对于kafkaBroker动态读取的Class，相信大部分人都还不怎么了解，因此分享这篇文章给大家参考一下，希望大家阅读完这篇文章后大有收获，下面让我们一起去了解一下吧！
企业建站必须是能够以充分展现企业形象为主要目的，是企业文化与产品对外扩展宣传的重要窗口，一个合格的网站不仅仅能为公司带来巨大的互联网上的收集和信息发布平台，创新互联面向各种领域：高空作业车租赁等成都网站设计、成都营销网站建设解决方案、网站设计等建站排名服务。
实现一个对于kafkaBroker 动态读取的Class - DynamicBrokersReader
DynamicBrokersReader
package com.mixbox.storm.kafka;
import backtype.storm.Config;
import backtype.storm.utils.Utils;
import com.mixbox.storm.kafka.trident.GlobalPartitionInformation;
import com.netflix.curator.framework.CuratorFramework;
import com.netflix.curator.framework.CuratorFrameworkFactory;
import com.netflix.curator.retry.RetryNTimes;
import org.json.simple.JSONValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Map;
/**
* 动态的Broker读 我们维护了有一个与zk之间的连接,提供了获取指定topic的每一个partition正在活动着的leader所对应的broker
* 这样你有能力知道，当前的这些topic，哪一些broker是活动的 * @author Yin Shuai
*/
public class DynamicBrokersReader {
public static final Logger LOG = LoggerFactory
.getLogger(DynamicBrokersReader.class);
// 对于Client CuratorFrameWork的封装
private CuratorFramework _curator;
// 在Zk上注册的位置
private String _zkPath;
// 指定的_topic
private String _topic;
public DynamicBrokersReader(Map conf, String zkStr, String zkPath,
String topic) {
_zkPath = zkPath;
_topic = topic;
try {
_curator = CuratorFrameworkFactory
.newClient(
zkStr,
Utils.getInt(conf
.get(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT)),
15000,
new RetryNTimes(
Utils.getInt(conf
.get(Config.STORM_ZOOKEEPER_RETRY_TIMES)),
Utils.getInt(conf
.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL))));
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
_curator.start();
}
public DynamicBrokersReader(String zkPath) {
this._zkPath = zkPath;
}
/**
* 确定指定topic下，每一个partition的leader，所对应的 主机和端口， 并将它们存入到全部分区信息中
*
*/
public GlobalPartitionInformation getBrokerInfo() {
GlobalPartitionInformation globalPartitionInformation = new GlobalPartitionInformation();
try {
// 拿到当前的分区数目
int numPartitionsForTopic = getNumPartitions();
/**
* /brokers/ids
*/
String brokerInfoPath = brokerPath();
// 默认的我们的分区数目就只有 0， 1 两个
for (int partition = 0; partition < numPartitionsForTopic; partition++) {
// 这里请主要参考分区和领导者的关系
int leader = getLeaderFor(partition);
// 拿到领导者以后的zookeeper路径
String leaderPath = brokerInfoPath + "/" + leader;
try {
byte[] brokerData = _curator.getData().forPath(leaderPath);
/**
* 在这里， 我们拿到的brokerData为：
* {"jmx_port":-1,"timestamp":"1403076810435"
* ,"host":"192.168.50.207","version":1,"port":9092} 注意
* 这里是字节数组开始转json
*/
Broker hp = getBrokerHost(brokerData);
/**
* 记录好 每一个分区 partition 所对应的 Broker
*/
globalPartitionInformation.addPartition(partition, hp);
} catch (org.apache.zookeeper.KeeperException.NoNodeException e) {
LOG.error("Node {} does not exist ", leaderPath);
}
}
} catch (Exception e) {
throw new RuntimeException(e);
}
LOG.info("Read partition info from zookeeper: "
+ globalPartitionInformation);
return globalPartitionInformation;
}
/**
* @return 拿到指定topic下的分区数目
*/
private int getNumPartitions() {
try {
String topicBrokersPath = partitionPath();
List children = _curator.getChildren().forPath(
topicBrokersPath);
return children.size();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* @return 拿到的topic在zookeeper注册的分区地址
* brokers/topics/storm-sentence/partitions
*/
public String partitionPath() {
return _zkPath + "/topics/" + _topic + "/partitions";
}
/**
* 持有的是Broker节点的id号码，这个id号是在配置的过程中为每一个Broker分配的
* @return /brokers/ids
*/
public String brokerPath() {
return _zkPath + "/ids";
}
/**
* get /brokers/topics/distributedTopic/partitions/1/state {
* "controller_epoch":4, "isr":[ 1, 0 ], "leader":1, "leader_epoch":1,
* "version":1 }
*
* 说明一下，在kafka之中，每一个分区都会有一个Leader，有0个或者多个的followers， 一个leader会处理这个分区的所有请求
* @param partition
* @return
*/
private int getLeaderFor(long partition) {
try {
String topicBrokersPath = partitionPath();
byte[] hostPortData = _curator.getData().forPath(
topicBrokersPath + "/" + partition + "/state");
@SuppressWarnings("unchecked")
Map value = (Map) JSONValue
.parse(new String(hostPortData, "UTF-8"));
Integer leader = ((Number) value.get("leader")).intValue();
return leader;
} catch (Exception e) {
throw new RuntimeException(e);
}
}
public void close() {
_curator.close();
}
/**
* [zk: localhost:2181(CONNECTED) 56] get /brokers/ids/0 {
* "host":"localhost", "jmx_port":9999, "port":9092, "version":1 }
*
*
* @param contents
* @return
*/
private Broker getBrokerHost(byte[] contents) {
try {
@SuppressWarnings("unchecked")
Map value = (Map) JSONValue
.parse(new String(contents, "UTF-8"));
String host = (String) value.get("host");
Integer port = ((Long) value.get("port")).intValue();
return new Broker(host, port);
} catch (UnsupportedEncodingException e) {
throw new RuntimeException(e);
}
}
} 对于以上代码须知：
1 ： 我们持有了一个ZkPath ， 在Storm-kafka的class之中我们默认的是/brokers
2 ： _topic , 目前我们是针对的是Topic， 也就是说我们的partition，leader都是针对于单个Topic的
3：
1 int numPartitionsForTopic = getNumPartitions();
针对与一个Topic，首先我们要取当前的分区数，一般的情况，我们在kafka之中默认的分区数为2
2 String brokerInfoPath = brokerPath();
拿到 /brokers/ids 的分区号
3： for (int partition = 0; partition < numPartitionsForTopic; partition++) {依次的遍历每一个分区
4：int leader = getLeaderFor(partition);String leaderPath = brokerInfoPath + "/" + leader;byte[] brokerData = _curator.getData().forPath(leaderPath);
再通过分区拿到领导者，以及领导者的路径，最后拿到领导者的数据：
我们举一个小例子
* 在这里， 我们拿到的brokerData为：
* {"jmx_port":-1,"timestamp":"1403076810435"
* ,"host":"192.168.50.207","version":1,"port":9092}
4：Broker hp = getBrokerHost(brokerData);
拿到某一个Topic自己的分区在kafka所对应的Broker，并且其封装到 globalPartitionInformation
5 globalPartitionInformation.addPartition(partition, hp);
GlobalPartitionInformaton底层维护了一个HashMap
简单的来说：DynamicBrokersReader 针对某一个Topic维护了 每一个分区 partition 所对应的 Broker
以上是“Storm-kafka中如何实现一个对于kafkaBroker动态读取的Class”这篇文章的所有内容，感谢各位的阅读！相信大家都有了一定的了解，希望分享的内容对大家有所帮助，如果还想学习更多知识，欢迎关注创新互联行业资讯频道！
