Storm-kafka接口怎么實(shí)現(xiàn)

本篇內(nèi)容主要講解“Storm-kafka接口怎么實(shí)現(xiàn)”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“Storm-kafka接口怎么實(shí)現(xiàn)”吧!

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

閱讀目的：了解Storm 如何來(lái)封裝kafka接口，如何處理Connection連接的封裝性問(wèn)題

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
 * 動(dòng)態(tài)的【分區(qū)連接】
 * @author Yin Shuai
 */

public class DynamicPartitionConnections {

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

	/**
	 * 持有了一個(gè) kafka底層的SimpleConsumer對(duì)象
	 * 持有了  具體的分區(qū)
	 * 
	 * @author Yin Shuai
	 */
	
	
	static class ConnectionInfo {

		//內(nèi)部維持了一個(gè)SimpleConsumer
		SimpleConsumer consumer;
		
		//分區(qū)
		Set<Integer> partitions = new HashSet();

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

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

	// kafkaConfig
	KafkaConfig _config;

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

	IBrokerReader _reader;

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

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

		/**
		 * 依據(jù)你所擁有的partition號(hào)，拿到你所對(duì)應(yīng)的Broker
		 * GlobalPartitionInformation中有Map<Integer, Broker>
		 * partitionMap，記錄了分區(qū)號(hào)與Broker所對(duì)應(yīng)的關(guān)系
		 */

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

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

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


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

	public SimpleConsumer getConnection(Partition partition) {

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

	/**
	 * @param port    固定的Broker
	 * @param partition  固定的分區(qū)
	 */
	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();
	}
}

     與前文有關(guān)

                1：       在DynamicPartitionConnections之中，我們持有了一個(gè) IBrokerReader的接口對(duì)象。

                2 ：       由于IBrokerReader 派生出了  

                                                    2.1 StaticBrokerReader

                                                    2.2 ZBrokerReader

       在這個(gè)序列的一系列博文之中，ZBrokerReader已經(jīng)進(jìn)行了詳盡的分析，并且在賦值的過(guò)程之中，IBrokerReader也是實(shí)例化為ZBrokerReader了。

     內(nèi)部類：

                DynamicPartitionConnections 持有了一個(gè) CinnectionInfo的內(nèi)部類

static class ConnectionInfo {

		//內(nèi)部維持了一個(gè)SimpleConsumer
		SimpleConsumer consumer;
		
		//分區(qū)
		Set<Integer> partitions = new HashSet();

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

      1：  對(duì)于每一個(gè)Connection內(nèi)部都維持了一個(gè)SimpleConsumer ，以及一個(gè) Set集合 partitions

      2 ：在DynamicPartitionConnections里面我們維持了一個(gè)_connections的對(duì)象

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

      3 ：在連接維護(hù)之中，關(guān)鍵的地方是維護(hù)一個(gè) register注冊(cè)的行為：

public SimpleConsumer register(Broker host, int partition) {

     4： 如果_connections之中沒(méi)有包含Broker，那么將會(huì)再建立一個(gè)新的連接，并且將Broker和Connection 注冊(cè)到_connections之中

    5：在注冊(cè)的過(guò)程之中，不包含就注冊(cè)，最后都直接取出SimpleConsumer，這個(gè)SimpleConsumer

封裝了

new ConnectionInfo(new SimpleConsumer(

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

_config.bufferSizeBytes, _config.clientId)）：

到此，相信大家對(duì)“Storm-kafka接口怎么實(shí)現(xiàn)”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！