Kafka教程(万字图文全面详解)

Kafka定义

Apache Kafka是分布式发布-订阅消息系统，它最初由LinkedIn公司开发，之后成为Apache顶级项目。

Kafka是一个由Scala和Java编写的企业级的消息发布和订阅系统，主要应用场景有：日志收集系统和消息系统。

Kafka作用

1.日志收集

一个公司可以用Kafka可以收集各种服务的log。

2.消息系统

解耦和生产者和消费者，这也是目前使用比较多的场景，相当于起了一个缓存消息的作用。

3.用户活动跟踪

Kafka经常被用来记录web用户，或者app用户的各种活动，如浏览网页、搜索、点击等活动。

4.运营指标

Kafka也经常用来记录运营监控数据，包括：收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

5.流式处理

比如spark streaming和storm等。

Kafka特点

1.高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒。

2.可扩展性：kafka集群支持热扩展；

3.持久性&可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失；

4.高并发：支持数千个客户端同时读写。

Kafka架构

它的架构包括以下组件：

1.话题（Topic）

消息的类别，主要用于对消息进行逻辑上的区分，每条发送到Kafka集群的消息都需要有一个指定的Topic，消费者根据Topic对指定的消息进行消费。

Kafka的数据就保存在Topic上，在每个 broker 上都可以创建多个 Topic。

2.分区 (Partition)

Topic的分区，每个 topic 可以有多个分区，分区的作用是做负载，提高 kafka 的吞吐量。

3.生产者（Producer）

Producer 即消息的生产者，负责往Kafka集群中发送消息，是消息的入口。

4.服务代理（Broker）

Kafka集群中的服务实例，也称之为节点，每个Kafka集群包含一个或者多个Broker。

简单的理解就是一台 Kafka服务器，Kafka Ccluster表示集群的意思，也被称为代理（Broker）。

5.消费者（Consumer）

消息的消费者，主动从Kafka集群中拉取消息，比如：从Broker拉数据，从而消费这些已发布的消息。

Kafka下载

Kafka官方下载地址：https://kafka.apache.org/downloads

根据自己的需求下载相应版本的kafka，kafka 是由scala语言编写，下载稳定版本，即前一个版本。

Kafka安装

Kafka下载好后，下面就开始Kafka安装了。

1.新建Kafka安装目录

命令：

cd /

cd /opt

mkdir kafka

2.Kafka解压安装包

tar -xzvf 解压包名称

比如，解压刚才下载的文件：

[root@mikechen kafka]# tar -zxvf kafka_2.12-3.3.1.tgz

然后再查看解压文件：

[root@mikechen kafka]# cd kafka_2.12-3.3.1.tgz
[root@mikechen kafka_2.12-3.3.1.tgz
bin  config  libs  LICENSE  licenses  NOTICE  site-docs

3.修改核心配置文件

进入config配置目录，修改配置文件 server.properties

比如：刚才我安装的目录在： /opt/kafka/config/server.properties

cd config

vi server.properties

修改如下6处内容：

vi /opt/kafka/config/server.properties

# 修改以下内容

#配置集群环境
broker.id=1
listeners=PLAINTEXT://server1:9092

#数据存放目录
log.dirs=/opt/kafka/data

#连接池地址信息
zookeeper.connect=server1:2181,server2:2181,server3:2181

# 文件尾部添加以下内容

#是否直接删除topic
delete.topic.enable=true

#主机名称
host.name=server1

# 创建数据存放目录
mkdir -p /opt/kafka/data
mkdir -p /opt/kafka/zkData
mkdir -p /opt/kafka/zkLog

修改核心配置说明：

• ​​broker.id​​ ：配置的是集群环境，要求每台kafka都有唯一的brokerid；
• log.dirs​​ ：数据存放的目录；
• zookeeper.connect​​ zookeeper：连接池地址信息；
• ​​delete.topic.enable​​ ：是否直接删除topic；
• host.name​​：主机名称
• ​​listeners：主要配置server名称，比如：PLAINTEXT://server1:9092

4.测试安装

1)启动zookeeper

kafka 是基于 Zookeeper 的消息管理系统，所以启动的时候是需要使用到 Zookeeper ，但其内置了Zookeeper ，所以只需要根据bin目录下的文件进行启动即可。

cd /opt/kafka/bin
./zookeeper-server-start.sh -daemon ../config/zookeeper.properties

Zookeeper 启动成功后会出现 binding to port 0.0.0.0/0.0.0.0:2181 所示结果表示启动成功。

2)再启动kafka

进入到kafka bin目录下， 启动如下指令

./kafka-server-start.sh -daemon ../config/server.propertie

这样Kafka就安装成功了。

Kafka命令

Kafka常用命令主要如下9种：

bin/kafka-server-start.sh -daemon config/server.properties

./kafka-server-stop.sh

3.创建Topic

命令：

bin/kafka-topics.sh --create --topic test0--zookeeper 127.0.0.1:2181

参数说明：

–topic：后面的test0是topic的名称；

–zookeeper：应该和server.properties文件中的zookeeper.connect一样；

4.查看所有的Topic列表

命令：

bin/kafka-topics.sh --list --zookeeper localhost:9092

5.查看所有的Topic的详细信息

命令：

bin/kafka-topics.sh --describe --zookeeper cdh-worker-1:2181/kafka

如果要查看单个 topic 信息：可在上述命令后面添加 –topic <topicName>。

6.查看Topic的分区和副本情况

命令：

bin/kafka-topics.sh --describe --zookeeper 127.0.0.1:2181 --topic test0

运行结果：

Topic:test0   PartitionCount:16       ReplicationFactor:3     Configs:
  
Topic: test0  Partition: 0    Leader: 0       Replicas: 0,2,1 Isr: 1,0,2

Topic: test0  Partition: 1    Leader: 1       Replicas: 1,2,0 Isr: 1,0,2

Topic: test0  Partition: 2    Leader: 2       Replicas: 2,0,1 Isr: 1,0,2

Topic: test0  Partition: 3    Leader: 1       Replicas: 1,2,0 Isr: 1,0,2

Topic: test0  Partition: 4    Leader: 2       Replicas: 2,0,1 Isr: 1,0,2

Topic: test0  Partition: 5    Leader: 0       Replicas: 0,1,2 Isr: 1,0,2

7.删除一个Topic

命令：

bin/kafka-topics.sh --delete --zookeeper 127.0.0.1:2181 --topic test0

8.Kafka发送消息

命令：

./kafka-console-producer.sh --broker-list localhost:9092 --topic test

9.Kafka接收消息

命令：加了–from-beginning 重头消费所有的消息

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

命令：不加–from-beginning 从最新的一条消息开始消费

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test

Kafka集群

Kafka集群搭建安装从本质上来讲，由一套多节点Zookeeper集群和一套多节点Kafka集群搭建组成。

如下图所示：

Kafka集群搭建共依赖三部分内容：jdk，zookeeper，kafka。

多节点安装Kafka比只需要创建多份配置文件，然后指定他们启动kafka服务即可。

1.下载Kafka

wget https://mirror.bit.edu.cn/apache/kafka/2.5.0/kafka_2.13-2.5.0.tgz

2.解压Kafka

tar -xvf kafka_2.13-2.5.0.tgz

3.修改配置文件

修改server.properties里面的broker.id，每台机器都不一样分别为：

broker.id=1

broker.id=2

broker.id=3

4.再修改里面的zookeeper的集群

zookeeper.connect=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181

5.依次启动kafka

bin/kafka-server-start.sh config/server.properties

Kafka原理

Kafka工作原理主要涉及消息生产、消息存储、以及消息消费等。

1.消息生产

我们将消息的发布（publish）称作 producer。

2.消费消费

将消息的订阅（subscribe）表述为 consumer。

3.broker(代理)

将中间的存储阵列称作 broker(代理)，这样就可以大致描绘出这样一个场面：

上图基本都是目前主流消息中间件的原型，可以很清楚的看见三者的关系。

生产者将数据生产出来，交给 broker 进行存储，如下图所示：

消费者需要消费数据了，就从broker中去拿出数据来，然后完成一系列对数据的处理操作。

备注：图上有个细节需要注意：producer 到 broker 的过程是 push也就是有数据就推送到 broker，而 consumer 到 broker 的过程是 pull是通过 consumer 主动去拉数据的。

多个 broker 协同合作，producer 和 consumer 部署在各个业务逻辑中被频繁的调用，三者通过 zookeeper管理协调请求和转发。

Kafka使用zookeeper作为其分布式协调框架，很好的将消息生产、消息存储、消息消费的过程结合在一起。

如下图所示：

这样一个高性能的分布式消息发布订阅系统就完成了。

Kafka应用

1.日志收集

一个公司可以用Kafka可以收集各种服务的log。

2.消息系统

解耦和生产者和消费者，这也是目前使用比较多的场景，相当于起了一个缓存消息的作用。

3.用户活动跟踪

Kafka经常被用来记录web用户，或者app用户的各种活动，如浏览网页、搜索、点击等活动。

4.运营指标

Kafka也经常用来记录运营监控数据，包括：收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

5.流式处理

比如spark streaming和storm等。

作者简介

陈睿|mikechen，10年+大厂架构经验，BAT资深面试官，就职于阿里巴巴、淘宝、百度等一线互联网大厂。

👇阅读更多mikechen架构文章👇

阿里架构 |双11秒杀 |分布式架构 |负载均衡 |单点登录 |微服务 |云原生 |高并发 |架构师

以上

关注作者「mikechen」公众号，获取更多技术干货！

后台回复【架构】，即可获取《阿里架构师进阶专题全部合集》，后台回复【面试】即可获取《史上最全阿里Java面试题总结》