Confulent-Kafka在Rainbond上的部署与使用

#1

Confulent-Kafka在Rainbond上的部署

Apache Kafka®的介绍

Apache Kafka®是分布式流处理平台，常用于建立可靠的实时流数据管道以在系统或应用时间传输数据，或者构建实时流应用程序，以转换或响应数据流。

Kafka在一个或多个可以跨越多个数据中心的服务器上作为集群运行。Kafka集群将记录流存储在被称为主题(Topic)的类别中。每个记录由一个键，一个值和一个时间戳组成。

kafka-apis
kafka-apis1402×850 90.3 KB

Kafka具有一下核心的API：

Producer API

应用程序可以将记录流发布到一个或多个Kafka主题。

Consumer API

应用程序可以订阅主题并处理为其生成的记录流。

Streams API

应用程序可以充当流处理器，使用一个或多个主题的输入流，并生成一个或多个输出主题的输出流，从而有效地将输入流转换为输出流。

Connector API

建立并运行可重用的生产者或使用者，将Kafka主题连接到现有应用程序或数据系统。例如，关系数据库的连接器可能会捕获对表的所有更改。

在Kafka中，客户端和服务器之间的通信是通过简单，高性能，与语言无关的TCP协议完成的。该协议已版本化，并与旧版本保持向后兼容性。Java客户端是为Kafka提供的，但是客户端支持多种语言。

主题和日志

Kafka为记录流提供的核心抽象是主题(Topic)。

主题是将记录发布到的类别或订阅源名称。Kafka中的主题始终是多用户的。这意味着一个主题可以有零个，一个或多个消费者来订阅写入该主题的数据。

对于每个主题，Kafka集群都会维护一个分区日志，如下所示：

log_anatomy

每个分区都是有序的，不可变的记录序列，这些记录连续地附加到结构化的提交日志中。分区中的每个记录均分配有一个顺序ID号，称为offset，它唯一标识分区中的每个记录。

无论是否已使用可配置的保留期限来使用它们，Kafka群集都会持久保留所有已发布的记录。例如，如果保留策略设置为两天，则在发布记录后的两天内，该记录可供使用，然后在两天后将其丢弃以释放空间。Kafka的性能相对于数据大小实际上是恒定的，这意味着长时间存储数据不是问题。

log_consumer
log_consumer1773×1080 103 KB

以上来自 https://docs.confluent.io/current/kafka/introduction.html 的翻译。

基于应用市场的部署

Confluent-Kafka 已经发布到了好雨公有应用市场，可以从应用市场一键安装：

kafka-3
kafka-32103×1080 144 KB

如何配置Confluent-Kafka

通用配置方式

Confluent-Kafka支持基于环境变量配置自身。配置的格式为：以KAFKA_开头的环境变量，并以下划线 _ 分割每个关键词， _ 会被替换为 .。例如：

声明环境变量 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1，将为kafka添加配置：offsets.topic.replication.factor =1。

集群内部访问

当 Broker 仅需要为 Rainbond 纳管下的同一个集群中的其他 Client 服务时，这是一种默认的情形。请 不要 再为 Kafka01 和 Kafka02 声明以下环境变量：

KAFKA_BROKER_ID

KAFKA_ADVERTISED_LISTENERS

上述两个环境变量，会由平台生成默认值，自动完成向zookeeper的注册，以及集群ID的声明；Broker 会默认监听 POD IP 的 9092 端口。

Client 仅需要依赖应用中的 zookeeper-cluster 即可通过 127.0.0.1:2181 这个地址，调取所有 Broker 的可用地址。

通过网关访问

当 Broker 需要通过网关向集群之外的其他 Client 暴露服务时，请为 Kafka01 和 Kafka02 声明以下环境变量：

KAFKA_LISTENERS = PLAINTEXT://${POD_IP}:9092

KAFKA_ADVERTISED_LISTENERS = PLAINTEXT://<网关IP>:<对外服务端口>

并开放 9092 端口对外服务。以下是一个示例：

kafka-4
kafka-42823×1080

kafka-5
kafka-52878×880

应用中的 zookeeper-cluster 需要开启 2181 的对外服务。Client 需要访问该对外服务的地址，即可调取所有 Broker 通过网关暴露的可用地址。

制作流程

代码地址为：https://github.com/goodrain-apps/cp-docker-images-rainbond

代码目录位于：debian/kafka

代码分支采用： 5.0.0-post

应用制作过程类似于Confluent-Zookeeper，制作文档参见https://t.goodrain.com/t/topic/1346

接下来讲解如何解析所依赖的Zookeeper地址。

详细代码位于：debian/kafka/include/etc/confluent/docker/rainbondenv 的63-70行

#detect ZooKeeper cluster 
if [ ! -n ${DEPEND_SERVICE-} ]; then
   echo "there is no zookeepers on my back... exiting"
   exit 1
else    
   zks=$(nslookup ${DEPEND_SERVICE%:*} | grep Address | sed '1d' | awk '{print $2":2181"}')
   export KAFKA_ZOOKEEPER_CONNECT=$(echo $zks | tr ' ' ',')
fi

特别声明一种用法，即平台默认生成的环境变量 ${DEPEND_SERVICE} ，该环境变量经过如代码中的处理并使用 nslookup 命令解析后，将返回Kafka依赖的所有zookeeper实例的真实IP地址。这种解析方法适用于需要进行客户端负载均衡的情况。

特别强调，该方式中，平台依赖关系只负责提供Zookeeper地址解析的信息来源。Kafka与Zookeeper之间的通信将直接通过Rainbond平台SDN内网实现，不再通过ServiceMesh代理。故而，这将是没有任何限制的TCP/IP通信，不再适用于各种网络治理插件的治理。
如果用户对于网络治理插件有需求，请直接声明KAFKA_ZOOKEEPER_CONNECT=127.0.0.1:2181来取消客户端负载均衡的设置。

Kafka-manager

同时发布的还包括一款图形化Kafka管理组件—— Kafka-manager。访问WEB界面后，声明zookeeper访问地址为 127.0.0.1:2181，即可从zookeeper中或取Kafka的brokers信息。

kafka-2
kafka-22013×1080 88.7 KB