理论

• 消息队列
  • 异步(两个不用同时在线)
  • 削峰

点对点模式

发布订阅模式

消费者拉取(kafka)(缺点：长轮询)(优点：消费者决定速率) 消息队列推送

==在此之前,要安装好jdk和Zookeeper集群== 安装位置:我猜是和Zookeeper安装的机器一样

1.下载安装

• 在官网下载(最新版本?)(一般不会出现版本兼容问题): kafka_2.12-2.4.1.tgz
• 上传到虚拟机
• 解压安装: tar -zxvf /soft/software/kafka_2.12-2.4.1.tgz -C /soft/module/
• 重命名: mv kafka_2.12-2.4.1 kafka

2.修改配置文件

1. 在kafka目录下新建文件夹: mkdir logs
2. 修改配置文件: cd config/ vim server.properties

#broker 的全局唯一编号，不能重复
broker.id=0
#删除 topic 功能使能彻底删除
delete.topic.enable=true
#kafka 运行日志存放的路径
log.dirs=/soft/module/kafka/logs
#配置连接Zookeeper 集群地址
zookeeper.connect=hadoop101:2181,hadoop102:2181,hadoop103:2181

#后面不是必须配置的内容
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘 IO 的线程数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400 
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400 
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#topic 在当前 broker 上的分区个数
num.partitions=1
#用来恢复和清理 data 下数据的线程数量
num.recovery.threads.per.data.dir=1 
#segment 文件保留的最长时间，超时将被删除
log.retention.hours=168

3. 配置环境变量 sudo vim /etc/profile

#KAFKA_HOME 环境变量配置
export KAFKA_HOME=/soft/module/kafka export PATH=$PATH:$KAFKA_HOME/bin

source /etc/profile 4. 分发文件 5. 修改集群其它机器kafka配置 分别在hadoop103和hadoop104上修改配置文件/opt/module/kafka/config/server.properties中的broker.id=1、broker.id=2 注：broker.id不得重复

3.启动集群

先启动Zookeeper

1. 单机启动 启动: bin/kafka-server-start.sh -daemon config/server.properties 关闭: bin/kafka-server-stop.sh stop
2. 集群启动 启动脚本

for i in hadoop101 hadoop102 hadoop103 do
echo "========== $i =========="
ssh	$i	'/soft/module/kafka/bin/kafka-server-start.sh
/soft/module/kafka/config/server.properties' done

4.kafka命令行操作

1. 查看当前服务器中的所有 topic bin/kafka-topics.sh --zookeeper hadoop1:2181 --list

2. 创建 topic bin/kafka-topics.sh --zookeeper hadoop1:2181 --create --replication-factor 3 --partitions 1 --topic first --topic 定义 topic 名 --replication-factor 定义副本数 --partitions 定义分区数

3. 删除 topic bin/kafka-topics.sh --zookeeper hadoop1:2181 --delete --topic first 需要 server.properties 中设置 delete.topic.enable=true 否则只是标记删除。

4. 发送消息

bin/kafka-console-producer.sh --broker-list hadoop1:9092 --topic first
>hello world
>atguigu	atguigu

5. 消费消息

bin/kafka-console-consumer.sh \
--zookeeper hadoop1:2181 --topic first

bin/kafka-console-consumer.sh \
--bootstrap-server hadoop1:9092 --topic first

bin/kafka-console-consumer.sh \
--bootstrap-server hadoop1:9092 --from-beginning --topic first

6. 查看某个Topic 的详情 bin/kafka-topics.sh --zookeeper hadoop1:2181 --describe --topic first
7. 修改分区数 bin/kafka-topics.sh --zookeeper hadoop1:2181 --alter --topic first --partitions 6