本篇內(nèi)容主要講解“Storm-kafka接口怎么實現(xiàn)”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學(xué)習(xí)“Storm-kafka接口怎么實現(xiàn)”吧!
10年積累的成都網(wǎng)站制作、成都做網(wǎng)站、外貿(mào)營銷網(wǎng)站建設(shè)經(jīng)驗，可以快速應(yīng)對客戶對網(wǎng)站的新想法和需求。提供各種問題對應(yīng)的解決方案。讓選擇我們的客戶得到更好、更有力的網(wǎng)絡(luò)服務(wù)。我雖然不認(rèn)識你，你也不認(rèn)識我。但先網(wǎng)站制作后付款的網(wǎng)站建設(shè)流程，更有碾子山免費網(wǎng)站建設(shè)讓你可以放心的選擇與我們合作。
閱讀背景： 如有需要，盡情參看本空間的另外一篇文檔
閱讀目的：了解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
* 動態(tài)的【分區(qū)連接】
* @author Yin Shuai
*/
public class DynamicPartitionConnections {
public static final Logger LOG = LoggerFactory
.getLogger(DynamicPartitionConnections.class);
/**
* 持有了一個 kafka底層的SimpleConsumer對象
* 持有了 具體的分區(qū)
*
* @author Yin Shuai
*/
static class ConnectionInfo {
//內(nèi)部維持了一個SimpleConsumer
SimpleConsumer consumer;
//分區(qū)
Set partitions = new HashSet();
public ConnectionInfo(SimpleConsumer consumer) {
this.consumer = consumer;
}
}
/**
* 也就是kafka的每一個節(jié)點都維持了一個COnnectionInfo,ConnectionInfo
*/
Map _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號，拿到你所對應(yīng)的Broker
* GlobalPartitionInformation中有Map
* partitionMap，記錄了分區(qū)號與Broker所對應(yīng)的關(guān)系
*/
Broker broker = _reader.getCurrentBrokers().getBrokerFor(
partition.partition);
return register(broker, partition.partition);
}
/**
* @param host
* 主機
* @param partition
* 分區(qū)
* @return 底層的SimpleConsumer 對象,這里存在一個注冊的行為，將主機和端口【broker】，和分區(qū)【partition】 注冊到 connections連接之中
*/
public SimpleConsumer register(Broker host, int partition) {
// Map _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)));
}
// --------- 在這里，不管之前有沒有都只取一次 -------------
//當(dāng)包含了，那就直接取出
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 固定的分區(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之中，我們持有了一個 IBrokerReader的接口對象。
2 ： 由于IBrokerReader 派生出了
2.1 StaticBrokerReader
2.2 ZBrokerReader
在這個序列的一系列博文之中，ZBrokerReader已經(jīng)進(jìn)行了詳盡的分析，并且在賦值的過程之中，IBrokerReader也是實例化為ZBrokerReader了。
內(nèi)部類：
DynamicPartitionConnections 持有了一個 CinnectionInfo的內(nèi)部類
static class ConnectionInfo {
//內(nèi)部維持了一個SimpleConsumer
SimpleConsumer consumer;
//分區(qū)
Set partitions = new HashSet();
public ConnectionInfo(SimpleConsumer consumer) {
this.consumer = consumer;
}
} 1： 對于每一個Connection內(nèi)部都維持了一個SimpleConsumer ，以及一個 Set集合 partitions
2 ：在DynamicPartitionConnections里面我們維持了一個_connections的對象
