一. Kafka简介
在Kafka中，客户端和服务器使用一个简单、高性能、支持多语言的 TCP 协议
Apache Kafka 是 一个分布式流处理平台，具有如下三个特性：
可以让你发布和订阅流式的记录，类似MQ
可以储存流式的记录，并且有较好的容错性
可以在流式记录产生时就进行处理
基于上述三个特性，衍生两大应用场景：
在应用之间构造实时流数据管道，即MQ
-
构建实时流式应用程序，对这些流数据进行转换或者影响，即流处理
（
通过Kafka Connect、Kafka Streams以支持连接其他系统的数据(譬如Elasticsearch、Hadoop等)）
Kafka通过四类API，来实现上述应用场景：
Producer API ：允许一个应用程序发布一串流式的数据到一个或者多个Kafka topic。
Consumer API ：允许一个应用程序订阅一个或多个 topic ，并且对发布给他们的流式数据进行处理。
Streams API ：允许一个应用程序作为一个流处理器，消费一个或者多个topic产生的输入流，然后生产一个输出流到一个或多个topic中去，在输入输出流中进行有效的转换。
Connector API ：允许构建并运行可重用的生产者或者消费者，将Kafka topics连接到已存在的应用程序或者数据系统。比如，连接到一个关系型数据库，捕捉表（table）的所有变更内容。
二. Kafka架构
在一个Kafka集群中，基本由如下组成部分：
Broker：即一个kafka节点
Topic：即主题，用来承载消息【
对于每一个topic， Kafka集群都会维持一个分区日志;每个分区都是有序且顺序不可变的记录集，并且不断地追加到结构化的commit log文件】Partition：分区，用于主题分片存储【
一个topic下有多个partition，topic是一个逻辑概念，partition则是物理概念；一个topic下的partition可以分布在多个broker上，实现水平扩展与负载均衡】Producer：生产者，向主题发布消息的应用
Consumer：消费者，从主题订阅消息的应用
Consumer Group：消费者组，由多个消费者组成【
灵活应用Consumer Group，可以实现消息的指定消费】
三. 集群搭建
- Kakfa依赖ZooKeeper存储Broker、Topic等信息，因此需要安装zookeeper
- zookeeper集群使用 Raft 选举模式，故至少要三个节点
3.1 下载安装包
kafka安装包中自带了zookeeper
#创建应用目录
mkdir -p /usr/local/kafka
mkdir -p /usr/local/zookeeper
#创建Kafka数据目录
mkdir -p /usr/local/kafka/logs
mkdir -p /usr/local/zookeeper/logs
#创建并进入下载目录
mkdir ~/downloads
cd ~/downloads
#下载安装包（kafka_2.12-2.3.0.tgz 其中2.12是Scala编译器的版本，2.3.0才是Kafka的版本）
wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.3.0/kafka_2.12-2.3.0.tgz
#解压到应用目录
tar -zvxf kafka_2.12-2.3.0.tgz -C /usr/local/kafka
3.2 修改配置
- 修改zookeeper配置
#进入应用目录
cd /usr/local/kafka/kafka_2.12-2.3.0/
#修改配置文件
vi config/zookeeper.properties
########## 修改zookeeper配置文件 ##########
#修改为自定义的zookeeper数据目录
dataDir=/usr/local/zookeeper
#修改为自定义的zookeeper日志目录
dataLogDir=/usr/local/zookeeper/logs
# 端口
clientPort=2181
#注释掉
#maxClientCnxns=0
#设置连接参数，添加如下配置
tickTime=2000 #为zk的基本时间单元，毫秒
initLimit=10 #Leader-Follower初始通信时限 tickTime*10
syncLimit=5 #Leader-Follower同步通信时限 tickTime*5
#设置broker Id的服务地址
server.0=${节点1主机}:2888:3888
server.1=${节点2主机}:2888:3888
server.2=${节点3主机}:2888:3888
# zookeeper数据目录添加myid配置
# 进入应用目录
cd /usr/local/zookeeper/
# 参照"broker Id的服务地址"配置，若为${主机1}，则节点id为0
echo ${节点id} > myid
- 修改kafka配置
#进入应用目录
cd /usr/local/kafka/kafka_2.12-2.3.0/
#修改配置文件
vi config/server.properties
########## 修改kafka配置文件 ##########
#kafka 运行日志存放的路径
log.dirs=/usr/local/kafka/logs
# 配置连接 Zookeeper 集群地址
zookeeper.connect=${节点1主机}:2181,${节点2主机}:2181,${节点3主机}:2181
#broker的全局唯一编号，每个节点不能重复，一般为：0,1,2
broker.id=${节点编号}
######## Socket配置 ########
# 处理网络请求的线程数量，默认
num.network.threads=3
# 用来处理磁盘IO的线程数量，默认
num.io.threads=8
# 发送套接字的缓冲区大小，默认
socket.send.buffer.bytes=102400
# 接收套接字的缓冲区大小，默认
socket.receive.buffer.bytes=102400
# 请求套接字的缓冲区大小，默认
socket.request.max.bytes=104857600
######## 日志留存策略 ########
# segment文件保留的最长时间，超时将被删除，默认
log.retention.hours=168
# 滚动生成新的segment文件的最大时间，默认
log.roll.hours=168
3.3 启动服务
- 启动zookeeper服务
#进入zookeeper根目录
cd /usr/local/kafka/kafka_2.12-2.3.0
#启动
sh bin/zookeeper-server-start.sh config/zookeeper.properties &
- 启动kafka服务
#开放端口
firewall-cmd --add-port=9092/tcp --permanent
#重新加载防火墙配置
firewall-cmd --reload
#进入kafka根目录
cd /usr/local/kafka/kafka_2.12-2.3.0
#启动
sh bin/kafka-server-start.sh config/server.properties &
#启动成功输出示例(最后几行)
[2019-06-26 21:48:57,183] INFO Kafka commitId: fc1aaa116b661c8a (org.apache.kafka.common.utils.AppInfoParser)
[2019-06-26 21:48:57,183] INFO Kafka startTimeMs: 1561531737175 (org.apache.kafka.common.utils.AppInfoParser)
[2019-06-26 21:48:57,185] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
# 后台启动
sh bin/kafka-server-start.sh -daemon config/server.properties
3.4 测试
- 创建一个 topic
# 创建topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
# 查询topic列表
bin/kafka-topics.sh --list --zookeeper localhost:2181
- 发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
This is a message
This is another message
- 启动一个consumer
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
This is a message
This is another message
3.5 集群测试
- 创建带副本topic
# 创建节点
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
# 查询状态：describe命令
# leader: 每个节点随机选择的部分分区的领导者;负责给定分区所有读写操作的节点
# replicas：复制分区日志的节点列表
# isr：一组“同步”replicas，是replicas列表的子集，它活着并被指到leader
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
