Storm-kafka中如何封裝DynamicBrokerReader類

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

在細(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 是對于DynamicBrokersReader的一個簡單的封裝
 * @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();

		// 很簡單， 指定了你多長時間開始去刷新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();    最新的刷新時間為系統(tǒng)的當(dāng)前時間

    4 long refreshMillis 

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

    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;
	}

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

的間隔，就會再次的去跟新：

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

一次。

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

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