1.kafka基础介绍

Kafka 中消息是以 topic 进行分类的， 生产者生产消息，消费者消费消息，都是面向 topic的。
topic 是逻辑上的概念，而 partition 是物理上的概念，每个 partition 对应于一个 log 文件，该 log 文件中存储的就是 producer 生产的数据。 Producer 生产的数据会被不断追加到该log 文件末端，且每条数据都有自己的 offset。 消费者组中的每个消费者， 都会实时记录自己消费到了哪个 offset，以便出错恢复时，从上次的位置继续消费。

由于生产者生产的消息会不断追加到 log 文件末尾， 为防止 log 文件过大导致数据定位效率低下， Kafka 采取了分片和索引机制，将每个 partition 分为多个 segment。 每个 segment对应两个文件——“.index”文件和“.log”文件。 这些文件位于一个文件夹下， 该文件夹的命名规则为： topic 名称+分区序号。例如， first 这个 topic 有三个分区，则其对应的文件夹为 first-0,first-1,first-2。

00000000000000000000.index
00000000000000000000.log
00000000000000170410.index
00000000000000170410.log
00000000000000239430.index
00000000000000239430.log

index 和 log 文件以当前 segment 的第一条消息的 offset 命名。下图为 index 文件和 log文件的结构示意图。

“.index”文件存储大量的索引信息，“.log”文件存储大量的数据，索引文件中的元数据指向对应数据文件中 message 的物理偏移地址。

2.kafka2.11安装文件配置

2.1 单机配置

需要事先安装好zookeeper:可参考zookeeper安装详情
kafka2.11下载
链接: https://pan.baidu.com/s/1P1WVr-KCeoaGYkNkikSkIw
提取码: 3ryz

tar-zxvf kafka_2.11-2.0.0.tgz -C /opt   //解压
mv kafka_2.11-2.0.0/ kafka211/  //改名
cd /opt/kafka211/config
vi server.properties

添加：

broker.id=0  //机器节点编号  多台机器的编号不能重复  0就是只有一台的意思

//listeners 是kafka真正bind的地址
//advertised.listeners 是暴露给外部的listeners，如果没有设置，会用listeners

advertised.listeners=PLAINTEXT://192.168.116.60:9092  //ip地址以及端口

log.dirs=/opt/kafka211/kafka-logs  //kafka消息存储的文件夹  需要自己去目录下创建kafka-logs 

zookeeper.connect=192.168.116.60:2181  //zookeeper ip地址以及端口

#末尾添加：
 delete.topic.enable=true  //添加才可以删除topic

保存退出
创建 kafka-logs 文件夹

cd /opt/kafka
mkdir kafka-logs  //创建文件夹

环境变量配置：

vi /etc/profile
export KAFKA_HOME=/opt/kafka211
export PATH=$KAFKA_HOME/bin:$PATH

source /etc/profile 生效

2.2 集群配置

vi server.properties

添加：

broker.id=0  //机器节点编号  多台机器的编号不能重复  另外两台为1 和 2

//listeners 是kafka真正bind的地址
//advertised.listeners 是暴露给外部的listeners，如果没有设置，会用listeners

advertised.listeners=PLAINTEXT://192.168.116.60:9092  //ip地址以及端口  每台机子写自己的ip地址即可

log.dirs=/opt/kafka211/kafka-logs  //kafka消息存储的文件夹  需要自己去目录下创建kafka-logs  

 //zookeeper集群 ip地址以及端口 
zookeeper.connect=hadoop001:2181,hadoop002:2181,hadoop003:2181
#末尾添加：
 delete.topic.enable=true  //添加才可以删除topic

保存退出
创建 kafka-logs 文件夹

cd /opt/kafka
mkdir kafka-logs  //创建文件夹

将kafka211文件夹以及环境变量分发给另外两台机器

scp -r /opt/kafka211  hadoop002:$PWD
scp -r /opt/kafka211  hadoop003:$PWD
scp -r /etc/profile hadoop002:$PWD
scp -r /etc/profile hadoop003:$PWD

另外两台别忘了source /etc/profile

3.启动kafka

zkServer.sh start  //启动zookeeper
zkServer.sh status  //查看zookeeper 状态

kafka-server-start.sh server.properties  
kafka-server-start.sh -daemon server.properties  //后台启动 集群分别执行本条命令启动kafka服务

//创建软链接 方便找到server.properties文件
[root@hadoop001 opt]# ln -s /opt/kafka211/config/server.properties /opt/server.properties

kafka-server-start.sh -daemon /opt/server.properties  //使用软链接后台启动

4.测试kafka

//查看队列信息：
[root@hadoop001 opt]# kafka-topics.sh --zookeeper 192.168.116.60:2181 --list

//创建队列信息：
[root@hadoop001 opt]# kafka-topics.sh --create --zookeeper 192.168.116.60:2181 --topic kb09demo --partitions 1 --replication-factor 1

//描述队列信息：
[root@hadoop001 opt]# kafka-topics.sh --zookeeper 192.168.116.60:2181 --topic kb09demo --describe
//返回的信息
Topic:kb09demo  PartitionCount:1        ReplicationFactor:1     Configs:
        Topic: kb09demo Partition: 0    Leader: 0       Replicas: 0   Isr: 0

//创建topic 分区数：5  副本数：1
[root@hadoop001 opt]# kafka-topics.sh --create --zookeeper 192.168.116.60:2181 --topic kb09demo2 --partitions 5 --replication-factor 1
//返回的信息：
Created topic "kb09demo2".

[root@hadoop001 opt]# kafka-topics.sh --zookeeper 192.168.116.60:2181 --topic kb09demo2 --describe
//返回的信息：
Topic:kb09demo2 PartitionCount:5        ReplicationFactor:1     Configs:
        Topic: kb09demo2        Partition: 0    Leader: 0       Replicas: 0    Isr: 0
        Topic: kb09demo2        Partition: 1    Leader: 0       Replicas: 0    Isr: 0
        Topic: kb09demo2        Partition: 2    Leader: 0       Replicas: 0    Isr: 0
        Topic: kb09demo2        Partition: 3    Leader: 0       Replicas: 0    Isr: 0
        Topic: kb09demo2        Partition: 4    Leader: 0       Replicas: 0    Isr: 0

//brokers：机器数   Replication factor: 2 是副本数 
[root@hadoop001 opt]# kafka-topics.sh --create --zookeeper 192.168.116.60:2181 --topic kb09demo3 --partitions 5 --replication-factor 2
//返回的信息：  报错原因：副本数 2不能超过机器数 1
Error while executing topic command : Replication factor: 2 larger than available brokers: 1.
[2020-12-02 16:24:27,241] ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 2 larger than available brokers: 1.
 (kafka.admin.TopicCommand$)

//删除topic
[root@hadoop001 opt]# kafka-topics.sh --zookeeper 192.168.116.60:2181 --topic kb09demo --delete

//生产者发送消息 消费者接收信息
[root@hadoop001 opt]# kafka-console-producer.sh --topic kb09demo --broker-list 192.168.116.60:9092

[root@hadoop001 ~]# kafka-console-consumer.sh --topic kb09demo --bootstrap-server 192.168.116.60:9092 --from-beginning

生产者发送消息 消费者接收信息结果：

//查看消息
[root@hadoop001 ~]# kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.116.60:9092 --topic kb09two -time -1 --offsets 1
kb09two:0:66
kb09two:1:67
kb09two:2:67

//消费消息
[root@hadoop001 ~]# kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic kb09two --from-beginning