189 8069 5689
本篇内容主要讲解“Storm-kafka接口怎么实现”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“Storm-kafka接口怎么实现”吧!
10年积累的成都网站制作、成都做网站、外贸营销网站建设经验，可以快速应对客户对网站的新想法和需求。提供各种问题对应的解决方案。让选择我们的客户得到更好、更有力的网络服务。我虽然不认识你，你也不认识我。但先网站制作后付款的网站建设流程，更有碾子山免费网站建设让你可以放心的选择与我们合作。
阅读背景： 如有需要，尽情参看本空间的另外一篇文档
阅读目的：了解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 partitions = new HashSet();
public ConnectionInfo(SimpleConsumer consumer) {
this.consumer = consumer;
}
}
/**
* 也就是kafka的每一个节点都维持了一个COnnectionInfo,ConnectionInfo
*/
Map _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
* 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 _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 partitions = new HashSet();
public ConnectionInfo(SimpleConsumer consumer) {
this.consumer = consumer;
}
} 1： 对于每一个Connection内部都维持了一个SimpleConsumer ，以及一个 Set集合 partitions
2 ：在DynamicPartitionConnections里面我们维持了一个_connections的对象