這篇文章主要介紹Storm-kafka中如何封裝DynamicBrokerReader類，文中介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們一定要看完！

創(chuàng)新互聯(lián)建站主營(yíng)膠州網(wǎng)站建設(shè)的網(wǎng)絡(luò)公司,主營(yíng)網(wǎng)站建設(shè)方案,成都app軟件開發(fā),膠州h5小程序制作搭建,膠州網(wǎng)站營(yíng)銷推廣歡迎膠州等地區(qū)企業(yè)咨詢

在細(xì)節(jié)上把握 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 是對(duì)于DynamicBrokersReader的一個(gè)簡(jiǎn)單的封裝
 * @author Yin Shuai
 */
public class ZkBrokerReader implements IBrokerReader {

	public static final Logger LOG = LoggerFactory
			.getLogger(ZkBrokerReader.class);

	GlobalPartitionInformation cachedBrokers;
	
	
	DynamicBrokersReader reader;
	
	
	long lastRefreshTimeMs;

	
	long refreshMillis;

	/**
	 * 
	 * @param conf
	 * @param topic
	 *            指定topic的zkBrokerReader
	 * @param hosts
	 */

	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();

		// 很簡(jiǎn)單， 指定了你多長(zhǎng)時(shí)間開始去刷新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  是對(duì)于  DynamicBrokersReader的一個(gè)簡(jiǎn)單封裝，ZkBrokerReader之中持有2個(gè)主要的Class

    1 GlobalPartitionInformatio  cachedBroker；

    2 DynamicBrokersReader  reader；

    3 long lastRefreshTimeMs;   最新的刷新時(shí)間

lastRefreshTimeMs = System.currentTimeMillis();    最新的刷新時(shí)間為系統(tǒng)的當(dāng)前時(shí)間

    4 long refreshMillis 

refreshMillis = host.refreshFreqSecs * 1000L  設(shè)定刷新的毫秒數(shù)為

    5 

public GlobalPartitionInformation getCurrentBrokers() {
		long currTime = System.currentTimeMillis();

		// 很簡(jiǎn)單， 指定了你多長(zhǎng)時(shí)間開始去刷新Brokerlibiao
		if (currTime > lastRefreshTimeMs + refreshMillis) {
			LOG.info("brokers need refreshing because " + refreshMillis
					+ "ms have expired");
			cachedBrokers = reader.getBrokerInfo();
			lastRefreshTimeMs = currTime;
		}
		return cachedBrokers;
	}

   每一次調(diào)用getCurrentBrokers，首先會(huì)取System.currentTimeMillis 當(dāng)當(dāng)前的系統(tǒng)時(shí)間超過了 最早的刷新時(shí)間+刷新

的間隔，就會(huì)再次的去跟新：

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

一次。

                ZkBrokerReader是對(duì)于DynamicBrokerReader的一個(gè)封裝，DynamicBrokerReader的Dynamic性質(zhì)并不程序動(dòng)態(tài)的因數(shù)，而只是簡(jiǎn)單在讀取ZK數(shù)據(jù)的過程之中，Zk數(shù)據(jù)已經(jīng)動(dòng)態(tài)的發(fā)生變化？

以上是“Storm-kafka中如何封裝DynamicBrokerReader類”這篇文章的所有內(nèi)容，感謝各位的閱讀！希望分享的內(nèi)容對(duì)大家有幫助，更多相關(guān)知識(shí)，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！