Kafka 是一个消息系统, 原本开发自 LinkedIn, 用作 LinkedIn 的活动流(Activity Stream)和运营数据处理管道(Pipeline)的基础。现在它已被多家不同类型的公司 作为多种类型的数据管道和消息系统使用。

活动流数据是几乎所有站点在对其网站使用情况做报表时都要用到的数据中最常规的部分。活动数据包括页面访问量(Page View)、被查看内容方面的信息以及搜索情况等内容。这种数据通常的处理方式是先把各种活动以日志的形式写入某种文件, 然后周期性地对这些文件进行统计分析。运营数据指的是服务器的性能数据(CPU、IO 使用率、请求时间、服务日志等等数据)。运营数据的统计方法种类繁多。

1. Kafka 简介

Kafka 是一种分布式的, 基于发布 / 订阅的消息系统。主要设计目标如下:

• 以时间复杂度为 O(1) 的方式提供消息持久化能力, 即使对 TB 级以上数据也能保证常数时间复杂度的访问性能
• 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒 100K 条以上消息的传输
• 支持 Kafka Server 间的消息分区, 及分布式消费, 同时保证每个 Partition 内的消息顺序传输
• 同时支持离线数据处理和实时数据处理
• Scale out: 支持在线水平扩展。

1.1 安装

# 下载及解压Zookeeper, kafka
# http://archive.apache.org/dist/
wget https://mirrors.aliyun.com/apache/zookeeper/stable/apache-zookeeper-3.6.3-bin.tar.gz
tar -xvf apache-zookeeper-3.6.3-bin.tar.gz
# 修改配置文件zoo.cfg

wget https://mirrors.aliyun.com/apache/kafka/2.8.2/kafka_2.13-2.8.2.tgz
tar -xaf kafka_2.13-2.8.2.tgz

# 编辑配置文件server.properties

# 启动
export ZOOKEEPER_HOME=/usr/local/zookeeper-3.4.12
export KAFKA_HOME=/usr/local/kafka_2.11
export PATH=$PATH:$ZOOKEEPER_HOME/bin:$KAFKA_HOME/bin

zkServer.sh start
kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

2. Kafka 基础概念

2.1 概念一: 生产者与消费者

对于 Kafka 来说客户端有两种基本类型: 生产者(Producer)和消费者(Consumer)。除此之外, 还有用来做数据集成的 Kafka Connect API 和流式处理的 Kafka Streams 等高阶客户端, 但这些高阶客户端底层仍然是生产者和消费者API, 它们只不过是在上层做了封装。

2.2 主题(Topic)与分区(Partition)

在 Kafka 中, 消息以主题(Topic)来分类, 每一个主题都对应一个「消息队列」, 这有点儿类似于数据库中的表。但是如果我们把所有同类的消息都塞入到一个“中心”队列中, 势必缺少可伸缩性, 无论是生产者/消费者数目的增加, 还是消息数量的增加, 都可能耗尽系统的性能或存储。

2.3 Broker 和集群(Cluster)

一个 Kafka 服务器也称为 Broker, 它接受生产者发送的消息并存入磁盘; Broker 同时服务消费者拉取分区消息的请求, 返回目前已经提交的消息。使用特定的机器硬件, 一个 Broker 每秒可以处理成千上万的分区和百万量级的消息。(现在动不动就百万量级..我特地去查了一把, 好像确实集群的情况下吞吐量挺高的..摁..)

2.4 多集群

随着业务发展, 我们往往需要多集群, 通常处于下面几个原因:

• 基于数据的隔离
• 基于安全的隔离
• 多数据中心(容灾)

当构建多个数据中心时, 往往需要实现消息互通。举个例子, 假如用户修改了个人资料, 那么后续的请求无论被哪个数据中心处理, 这个更新需要反映出来。又或者, 多个数据中心的数据需要汇总到一个总控中心来做数据分析。

3. Kafka基本配置

3.1 服务端的配置文件(server.properties)

3.2 生产端的配置文件(producer.properties)

3.3 消费端的配置文件(consumer.properties)

3.4 用户认证

# 创建管理用户admin, 密码是admin
kafka-configs.sh --zookeeper 127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=admin],SCRAM-SHA-512=[password=admin]' --entity-type users --entity-name admin

# 创建生产者 iawen, 密码是 123456
kafka-configs.sh --zookeeper 127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=123456],SCRAM-SHA-512=[password=123456]' --entity-type users --entity-name iawen

kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:iawen --operation Write --topic test


# 创建消费者 rec_user, 密码是 123456
kafka-configs.sh --zookeeper 127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=123456],SCRAM-SHA-512=[password=123456]' --entity-type users --entity-name rec_user

# 赋Topic 和 group 权限给用户
kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:rec_user --operation Read --topic test

kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:reader --operation Read --group test-group

kafka-configs.sh --zookeeper 127.0.0.1:2181 --describe --entity-type users --entity-name iawen
kafka-acls.sh --authorizer-properties zookeeper.connect=127.0.0.1:2181 --list

修改 server.properties 里的配置:

# 启用ACL
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
# 设置admin为超级用户
super.users=User:admin
# 启用SCRAM机制, 采用SCRAM-SHA-512算法
sasl.enabled.mechanisms=SCRAM-SHA-512
# 为broker间通讯开启SCRAM机制, 采用SCRAM-SHA-512算法
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

security.inter.broker.protocol=SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://192.168.137.3:9092
advertised.listeners=SASL_PLAINTEXT://192.168.137.3:9092

auto.create.topics.enable=false

KAFKA_OPTS=-Djava.security.auth.login.config=$KAFKA_HOME/config/kafka-broker-jaas.conf kafka-server-start.sh $KAFKA_HOME/config/server.properties

kafka-broker-jaas.conf 文件内容如下:

KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="123456";
};

4. Kafka 命令

4.1 常用操作

export PATH=/usr/local/confluent/bin:$PATH

kafka-topics.sh --zookeeper 127.0.0.1:2181 --create --topic rec_behave --partitions 1 --replication-factor 1

kafka-topics.sh --zookeeper 127.0.0.1:2181 --list

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list test_behavior

kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name __consumer_offsets --describe
kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name __consumer_offsets --alter --delete-config cleanup.policy
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic __consumer_offsets --partition 0 --from-beginning --formatter 'kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter'

# 1296000000  --> 15 days, 2592000000  --> 30 days
kafka-topics.sh --zookeeper localhost:2181 -topic xxxxx --alter --config retention.ms=2592000000 #

# 查看与删除topic下的持久化文件命令;  因为kafka一直运行的话 硬盘占用越来越多; 
# kafka-topics -delete --zookeeper 127.0.0.1:2181 --topic 162btc
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group xxx --topic xxx --execute --reset-offsets --to-offset 0

4.2 SASL 模式下

• config/kafka-client-jaas.conf

KafkaClient {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="xxxxx";
};

• config/consumer-sasl.properties

security.protocol = SASL_PLAINTEXT
sasl.mechanism = SCRAM-SHA-512

• bin/kafka-consumer-groups-sasl.sh

ROOT=$(dirname $0)
export KAFKA_OPTS=" -Djava.security.auth.login.config=$ROOT/../config/kafka_client_jaas.conf"
exec $ROOT/kafka-run-class.sh kafka.admin.ConsumerGroupCommand "$@" --command-config $ROOT/../config/consumer-sasl.properties

kafka-consumer-groups-sasl.sh --bootstrap-server 127.0.0.1:9092 --list 
kafka-consumer-groups-sasl.sh --bootstrap-server 127.0.0.1:9092 --group rec-group --describe

5. 其他

5.1 线上Kafka突发rebalance异常

参考: https://www.cnblogs.com/chanshuyi/p/kafka_rebalance_quick_guide.html