Storm-kafka接口怎么實現

本篇內容主要講解“Storm-kafka接口怎么實現”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“Storm-kafka接口怎么實現”吧!

閱讀背景： 如有需要，盡情參看本空間的另外一篇文檔

閱讀目的：了解Storm 如何來封裝kafka接口，如何處理Connection連接的封裝性問題

package com.mixbox.storm.kafka;

import kafka.javaapi.consumer.SimpleConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.mixbox.storm.kafka.trident.IBrokerReader;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * 2014/07/22
 * 動態的【分區連接】
 * @author Yin Shuai
 */

public class DynamicPartitionConnections {

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

	/**
	 * 持有了一個 kafka底層的SimpleConsumer對象
	 * 持有了  具體的分區
	 * 
	 * @author Yin Shuai
	 */
	
	
	static class ConnectionInfo {

		//內部維持了一個SimpleConsumer
		SimpleConsumer consumer;
		
		//分區
		Set<Integer> partitions = new HashSet();

		public ConnectionInfo(SimpleConsumer consumer) {
			this.consumer = consumer;
		}
	}

	/**
	 * 也就是kafka的每一個節點都維持了一個COnnectionInfo,ConnectionInfo
	 */
	Map<Broker, ConnectionInfo> _connections = new HashMap();

	// kafkaConfig
	KafkaConfig _config;

	/**
	 * IBrokerReader 基本上 IbroerReader這里初始化的是ZkBrokerReader
	 */

	IBrokerReader _reader;

	/**
	 * @param config
	 *            kafka配置
	 * @param brokerReader
	 *            IBrokerReader-用于拿到當前的接口
	 */
	public DynamicPartitionConnections(KafkaConfig config,
			IBrokerReader brokerReader) {
		_config = config;
		_reader = brokerReader;
	}

	/**
	 * @param partition  分區
	 * @return
	 */
	public SimpleConsumer register(Partition partition) {

		/**
		 * 依據你所擁有的partition號，拿到你所對應的Broker
		 * GlobalPartitionInformation中有Map<Integer, Broker>
		 * partitionMap，記錄了分區號與Broker所對應的關系
		 */

		Broker broker = _reader.getCurrentBrokers().getBrokerFor(
				partition.partition);
		return register(broker, partition.partition);
	}

	/**
	 * @param host
	 *            主機
	 * @param partition
	 *            分區
	 * @return 底層的SimpleConsumer 對象,這里存在一個注冊的行為，將主機和端口【broker】，和分區【partition】 注冊到 connections連接之中
	 */
	public SimpleConsumer register(Broker host, int partition) {

		// Map<Broker, ConnectionInfo> _connections = new HashMap();


		//如果連接之中沒有包含了Broker，那么建立一個新的連接，并且將這個  主機和連接注冊到  _connections之中
		if (!_connections.containsKey(host)) {
			_connections.put(host, new ConnectionInfo(new SimpleConsumer(
					host.host, host.port, _config.socketTimeoutMs,
					_config.bufferSizeBytes, _config.clientId)));
		}
		
		// ---------   在這里，不管之前有沒有都只取一次 -------------
		
		//當包含了，那就直接取出
		ConnectionInfo info = _connections.get(host);
		info.partitions.add(partition);
		return info.consumer;
	}

	public SimpleConsumer getConnection(Partition partition) {

		// ConnectionInfo 之中封裝了一個simpleConsumer
		ConnectionInfo info = _connections.get(partition.host);
		if (info != null) {
			return info.consumer;
		}
		return null;
	}

	/**
	 * @param port    固定的Broker
	 * @param partition  固定的分區
	 */
	public void unregister(Broker port, int partition) {
		ConnectionInfo info = _connections.get(port);
		info.partitions.remove(partition);
		if (info.partitions.isEmpty()) {
			info.consumer.close();
			_connections.remove(port);
		}
	}

	public void unregister(Partition partition) {
		unregister(partition.host, partition.partition);
	}

	public void clear() {
		for (ConnectionInfo info : _connections.values()) {
			info.consumer.close();
		}
		_connections.clear();
	}
}

     與前文有關

                1：       在DynamicPartitionConnections之中，我們持有了一個 IBrokerReader的接口對象。

                2 ：       由于IBrokerReader 派生出了  

                                                    2.1 StaticBrokerReader

                                                    2.2 ZBrokerReader

       在這個序列的一系列博文之中，ZBrokerReader已經進行了詳盡的分析，并且在賦值的過程之中，IBrokerReader也是實例化為ZBrokerReader了。

     內部類：

                DynamicPartitionConnections 持有了一個 CinnectionInfo的內部類

static class ConnectionInfo {

		//內部維持了一個SimpleConsumer
		SimpleConsumer consumer;
		
		//分區
		Set<Integer> partitions = new HashSet();

		public ConnectionInfo(SimpleConsumer consumer) {
			this.consumer = consumer;
		}
	}

      1：  對于每一個Connection內部都維持了一個SimpleConsumer ，以及一個 Set集合 partitions

      2 ：在DynamicPartitionConnections里面我們維持了一個_connections的對象

Map<Broker, ConnectionInfo> _connections = new HashMap();

      3 ：在連接維護之中，關鍵的地方是維護一個 register注冊的行為：

public SimpleConsumer register(Broker host, int partition) {

     4： 如果_connections之中沒有包含Broker，那么將會再建立一個新的連接，并且將Broker和Connection 注冊到_connections之中

    5：在注冊的過程之中，不包含就注冊，最后都直接取出SimpleConsumer，這個SimpleConsumer

封裝了

new ConnectionInfo(new SimpleConsumer(

host.host, host.port, _config.socketTimeoutMs,

_config.bufferSizeBytes, _config.clientId)）：

到此，相信大家對“Storm-kafka接口怎么實現”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！