day09 ----- Kafka

Kafka 为什么那么快？
1）顺序读写
2）直接使用系统自身缓存
3）零拷贝（直接复制到NIC缓冲区，避免重新复制数据）
4）分区，分段，索引
5）批量处理，合并小的请求，以流的方式进行交互

Kafka 的整体架构？
zookeeper：分布式协调服务，协调kafka应用。
broker：kafka的节点。
producer：生产者生产数据
consumer：消费者消费数据。
consumer group ：消费者组，组内的消费者默认是不能重复消费。
topic：主题，是消息的归类。
partition：分区，将数据分为多份，进行存储。
replicas：副本，数据是冗余存储的。
segment：文件段，文件按照1个g大小进行拆分。主要包括两个文件.index索引和.log日志文件。

ack参数?

0：生产者只负责发送数据，不关心数据是否丢失，响应的状态码为0（丢失的数据，需要再次发送）
1：partition的leader收到数据，响应的状态码为1
-1：所有的从节点都收到数据，响应的状态码为-1

kafka 服务启动

# & 后台启动
cd /opt/servers/kafka_2.12-2.2.0
nohup bin/kafka-server-start.sh config/server.properties 2>&1 &

停止

cd /opt/servers/kafka_2.12-2.2.0
bin/kafka-server-stop.sh

实时数据处理

没有主从，通过zookepper 完成协调

cd /opt/softwares
tar -xvzf kafka_2.12-2.2.0.tgz -C ../servers/

hadoop01服务器修改kafka配置文件
hadoop01执行以下命令进入到kafka的配置文件目录，修改配置文件

hadoop01执行以下命令创建数据文件存放目录

mkdir -p  /opt/servers/kafka_2.12-2.2.0/logs 

cd /opt/servers/kafka_2.12-2.2.0/config

vim server.properties

修改以下内容

broker.id=0
log.dirs=/opt/servers/kafka_2.12-2.2.0/logs
zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181
delete.topic.enable=true # 默认关闭，需额外添加

hadoop01执行以下命令，将hadoop01服务器的kafka安装包发送到hadoop02和hadoop03服务器

cd /opt/servers/

scp -r kafka_2.12-2.2.0/ hadoop02:$PWD
scp -r kafka_2.12-2.2.0/ hadoop03:$PWD

hadoop02与hadoop03服务器修改kafka配置文件

hadoop02使用以下命令修改kafka配置文件

cd /opt/servers/kafka_2.12-2.2.0/config

vim server.properties

broker.id=1
log.dirs=/opt/servers/kafka_2.12-2.2.0/logs
zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181
delete.topic.enable=true

hadoop03使用以下命令修改kafka配置文件

cd /opt/servers/kafka_2.12-2.2.0/config

vim server.properties

broker.id=2
log.dirs=/opt/servers/kafka_2.12-2.2.0/logs
zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181
delete.topic.enable=true

注意事项：在kafka启动前，一定要让zookeeper启动起来。
可三台同时 ，执行以下命令将kafka进程启动在后台

cd /opt/servers/kafka_2.12-2.2.0

nohup bin/kafka-server-start.sh config/server.properties 2>&1 &

&：后台启动
2>&1： 会产生日志，有异常可在日志查看

后台启动

bin/kafka-topics.sh --create --topic test --bootstrap-server hadoop01:9092,hadoop02:9092,hadoop03:9092 --partitions 3 --replication-factor 2

bin/kafka-topics.sh --list --bootstrap-server hadoop01:9092

bin/kafka-console-producer.sh --topic test --broker-list hadoop01:9092,hadoop02:9092

生产数据保存到哪？

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

为什么这样接收数据

全局乱序，局部(分区)有序

bin/kafka-topics.sh --describe --topic test --bootstrap-server hadoop01:9092,hadoop02:9092,hadoop03:9092

leader 是由zookepper指定的
副本数 Replicas 两个指定的不同节点

cousumer offsets 0-50
消费者 ---- 偏移量

任意kafka服务器执行以下命令可以增加topic分区数

cd /opt/servers/kafka_2.12-2.2.0

bin/kafka-topics.sh --bootstrap-server hadoop01:9092 --alter --topic test --partitions 5

producer生产者基于topic主题生产消息

分区：

mmap 可以直接内存操作： 零拷贝
ack 确认队列

如何保证数据不丢失？
保证数据不丢失，自身有副本数机制，副本因子
从消费者角度：接收数据时，保存好offset 值即可