Storm

155次阅读

共计 2360 个字符，预计需要花费 6 分钟才能阅读完成。

这篇文章主要介绍 Storm-kafka 中如何封装 DynamicBrokerReader 类，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！

在细节上把握 DynamicBrokerReder 的封装类 – ZkBrokerReader

package com.mixbox.storm.kafka.trident;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.mixbox.storm.kafka.DynamicBrokersReader;
import com.mixbox.storm.kafka.ZkHosts;
import java.util.Map;
 * 2014/07/22
 *  在 ZK 中间拿到  GlobalPartitionInformation
 * 
 * ZkBrokerReader  是对于 DynamicBrokersReader 的一个简单的封装
 * @author Yin Shuai
 */
public class ZkBrokerReader implements IBrokerReader {
 public static final Logger LOG = LoggerFactory
 .getLogger(ZkBrokerReader.class);
 GlobalPartitionInformation cachedBrokers;

 public ZkBrokerReader(Map conf, String topic, ZkHosts hosts) {
 reader = new DynamicBrokersReader(conf, hosts.brokerZkStr,
 hosts.brokerZkPath, topic);
 cachedBrokers = reader.getBrokerInfo();
 lastRefreshTimeMs = System.currentTimeMillis();
 refreshMillis = hosts.refreshFreqSecs * 1000L;
 @Override
 public GlobalPartitionInformation getCurrentBrokers() {long currTime = System.currentTimeMillis();
 //  很简单，  指定了你多长时间开始去刷新 Brokerlibiao
 if (currTime   lastRefreshTimeMs + refreshMillis) {
 LOG.info( brokers need refreshing because   + refreshMillis
 +  ms have expired 
 cachedBrokers = reader.getBrokerInfo();
 lastRefreshTimeMs = currTime;
 return cachedBrokers;
 @Override
 public void close() {reader.close();
}

总览我们的 Code：

ZkBrokerReader 是对于 DynamicBrokersReader 的一个简单封装，ZkBrokerReader 之中持有 2 个主要的 Class

1 GlobalPartitionInformatio cachedBroker；

2 DynamicBrokersReader reader；

3 long lastRefreshTimeMs; 最新的刷新时间

lastRefreshTimeMs = System.currentTimeMillis();  最新的刷新时间为系统的当前时间

4 long refreshMillis

refreshMillis = host.refreshFreqSecs * 1000L  设定刷新的毫秒数为

5

public GlobalPartitionInformation getCurrentBrokers() {long currTime = System.currentTimeMillis();
 //  很简单，  指定了你多长时间开始去刷新 Brokerlibiao
 if (currTime   lastRefreshTimeMs + refreshMillis) {
 LOG.info( brokers need refreshing because   + refreshMillis
 +  ms have expired 
 cachedBrokers = reader.getBrokerInfo();
 lastRefreshTimeMs = currTime;
 return cachedBrokers;
 }

每一次调用 getCurrentBrokers，首先会取 System.currentTimeMillis 当当前的系统时间超过了最早的刷新时间 + 刷新

的间隔，就会再次的去跟新：

cachedBrokers = reader.getBrokerInfo()；getBrokerInfo() 方法每调用一次，也就重新在 zk 之中重新去取

一次。

ZkBrokerReader 是对于 DynamicBrokerReader 的一个封装，DynamicBrokerReader 的 Dynamic 性质并不程序动态的因数，而只是简单在读取 ZK 数据的过程之中，Zk 数据已经动态的发生变化？

以上是“Storm-kafka 中如何封装 DynamicBrokerReader 类”这篇文章的所有内容，感谢各位的阅读！希望分享的内容对大家有帮助，更多相关知识，欢迎关注丸趣 TV 行业资讯频道！

正文完