黄埔班第45节：kafka消息队列应用实战-1

# 由于kafka的服务协调是基于zookeeper实现的，所以要先下载安装zookeeper
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.7/apache-zookeeper-3.5.7-bin.tar.gz 

# 下面在单机上（不同端口）安装zookeeper集群（伪集群）
# 创建集群文件夹
mkdir zookeeper-cluster
# 1、解压，重命名
tar -zxvf apache-zookeeper-3.5.8-bin.tar.gz
[root@helloworld zookeeper-cluster]# mv apache-zookeeper-3.5.8 zookeeper1
# 提前把data和logs文件创建好
# 2、在conf里修改zoo_sample.cfg为zoo.cfg进行修改
[root@helloworld conf]# cp zoo_sample.cfg zoo.cfg
[root@helloworld conf]# vim zoo.cfg
dataDir=/usr/local/zookeeper-cluster/zookeeper1/data
dataLogDir=/usr/local/zookeeper-cluster/zookeeper1/logs
clientPort=2181 # 每个机器的端口不能一样
#maxClientCnxns=60  # increase this if you need to handle more clients,生产环境中注意一下这个参数
server.1=127.0.0.1:2881:3881
server.2=127.0.0.1:2882:3882
server.3=127.0.0.1:2883:3883  # 增加通信内容三台机器一样：server.服务编号=服务地址:LF通信端口:选举端口
# 在三台机器的data目录dataDir=/usr/local/zookeeper-cluster/zookeeper1/data
# data里创建文件myid，三台分别写入值1、2、3
cd /usr/local/zookeeper-cluster/zookeeper1/data
[root@helloworld data]# vim myid
1
# 3、复制
[root@helloworld zookeeper-cluster]# cp -r zookeeper1/ zookeeper2
[root@helloworld zookeeper-cluster]# cp -r zookeeper1/ zookeeper3

# 修改zoo.cfg和myid
cd /usr/local/zookeeper-cluster/zookeeper2/data
[root@helloworld data]# vim myid
2
cd /usr/local/zookeeper-cluster/zookeeper3/data
[root@helloworld data]# vim myid
3
# 4、创建的sh文件统一启动zookeeper，zookeeper-start.sh
[root@helloworld zookeeper-cluster]# vim zookeeper-start.sh
/usr/local/zookeeper-cluster/zookeeper1/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper2/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper3/bin/zkServer.sh start
# 批处理文件授权
chmod 777 zookeeper-start.sh
# 启动zookeeper集群
./zookeeper-start.sh

# 5、服务管理命令
zkServer.sh start|stop|restart|status
# 客户端连接,客户端关闭：quit ,推荐使用界面操作工具zooInspector，因为zookeeper的数据是树形结构存储的，使用界面会更为直观
zkCli.sh -server ${ip}:${port}

# 下载kafka
wget http://mirror.bit.edu.cn/apache/kafka/2.4.0/kafka_2.13-2.4.0.tgz
# 1、解压后移动到/usr/local下
tar -xvf  kafka_2.13-2.4.0.tgz
mv kafka_2.13-2.4.0 kakfa
mv kafka /usr/local/

# 2、进入config目录，修改server.properties
[root@helloworld config]# vim server.properties
# kafka的broker编号，必须是一个整数，服务编号，在集群中不能一样
broker.id=11
# topic允许删除的，如果你设置成false，删除topic对kafka是无效的，默认就是true
delete.topic.enable=true
port=9092
# 处理网络请求的线程数量
num.network.threads=3
# 磁盘IO的线程数量
num.io.threads=8
# 发送socket的缓冲区大小
socket.send.buffer.bytes=102400
# 接收socket的缓冲区大小
socket.receive.buffer.bytes=102400
# 请求socket的缓冲区大小
socket.request.max.bytes=104857600
# 是保存消息日志的文件目录,data目录要提前创建好,partition分区就存放在这个目录下
log.dirs=/usr/local/kafka/data
# topic在创建时默认多少个分区
num.partitions=1
# 我们的消息数据默认保存时间长度，单位是小时 168小时=7天，不是永久保存
# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168
# 一个log文件的大小，这里默认是1G，如果超过1G就再创建一个新文件
log.segment.bytes=1073741824
# zookeeper的集群地址
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183

# 启动服务命令，要指定配置文件，后台启动
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
# 关闭服务命令
/usr/local/kafka/bin/kafka-server-stop.sh /usr/local/kafka/config/server.properties

[root@helloworld local]# vim /etc/profile
export KAFKA_HOME=/opt/kafka
export PATH=$PATH:$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin:$KAFKA_HOME/bin
# 使配置生效(已连接的会话需要重新连接)
source /etc/profile
# 加入环境变量后，直接使用命令
kafka-server-start.sh

# 0.查看topics命令帮助
kafka-topics.sh --help
# 1.查看当前服务器中所有的topic，需要连接zookeeper节点
kafka-topics.sh --zookeeper 127.0.0.1:2181 --list
# 2.创建topic，--replication-factor
kafka-topics.sh --zookeeper  127.0.0.1:2181 --create --partitions 3 --replication-factor 1 --topic topicfirst
# Error while executing topic command : Replication factor: 2 larger than available brokers: 1.
--topic # topic名称
--partitions # 定义分区数,partition的命名规则<topicName-partiton序号>
--replication-factor # 副本数量,1代表就是当前主分区没有副本,2代表每个主分区有一个副本分区,副本分区和主分区不能在同一个broker
# topic创建后，进入log.dirs=/usr/local/kafka/data 存储目录，发现topicfirst的3个分区文件夹
[root@helloworld data]# ls
topicfirst-0  topicfirst-1  topicfirst-2
[root@helloworld data]# cd topicfirst-0
[root@helloworld topicfirst-0]# ls 
00000000000000000000.index  00000000000000000000.log  00000000000000000000.timeindex  leader-epoch-checkpoint
# 000000000是偏移量

# 3.删除topic
# 如果配置文件server.properties里delete.topic.enable设为false，删除是无效的
# Note: This will have no impact if delete.topic.enable is not set to true.
kafka-topics.sh --zookeeper 127.0.0.1:2181 --delete --topic topicfirst
# 4.查看topic的详情
kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic topicfirst
# 5.修改topic分区数,注意分区只能增加，不能减少，因为可能会破坏分区数据
kafka-topics.sh --zookeeper 127.0.0.1:2181 -alter --topic topicfirst --partitions 5

[root@helloworld bin]# ./kafka-console-producer.sh --help
# 发送消息,broker是集群多个的话，要用,隔开,如192.168.1.120:9092,192.168.1.130:9093,192.168.1.140:9092
kafka-console-producer.sh --topic topicfirst --broker-list 127.0.0.1:9092
# 接收消息
# --from-beginning 如果加了，会从topic头来接收消息（监听前已发送的消息已持久化，需要加该参数进行接收）
kafka-console-consumer.sh --topic topicfirst --from-beginning --bootstrap-server 127.0.0.1:9092

[root@helloworld data]# cd topicfirst-0
[root@helloworld topicfirst-0]# ls 
00000000000000000000.index  # 保存的消息的顺序编号
00000000000000000000.log    # 实际保存消息数据的

kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic topicfirst
Topic: topicfirst       PartitionCount: 5       ReplicationFactor: 1    Configs: 
Topic: topicfirst       Partition: 0    Leader: 11      Replicas: 11    Isr: 11
Topic: topicfirst       Partition: 1    Leader: 11      Replicas: 11    Isr: 11
Topic: topicfirst       Partition: 2    Leader: 11      Replicas: 11    Isr: 11
Topic: topicfirst       Partition: 3    Leader: 11      Replicas: 11    Isr: 11
Topic: topicfirst       Partition: 4    Leader: 11      Replicas: 11    Isr: 11

这里的11是broker的id，在config/server.properties 配置