1.zookeeper集群部署

1.1部署zookeeper三台机器

#官网下载zookeeper压缩包

#上传服务器
cd /opt

mkdir -p /data/ota_soft

tar zxvf zookeeper-3.4.12.tar.gz -C /data/ota_soft/

ln -sf /data/ota_soft/zookeeper-3.4.12 /data/ota_soft/zookeeper

mkdir /data/ota_soft/zookeeper/{data,logs}

#修改配置文件
cd /data/ota_soft/zookeeper/conf/
cp zoo_sample.cfg zoo.cfg
vim zoo.cfg
 
tickTime=2000
initLimit=10
syncLimit=5
dataLogDir=/data/ota_soft/zookeeper/logs
dataDir=/data/ota_soft/zookeeper/data
clientPort=2181
server.1=master1:2888:3888
server.2=slave1:2888:3888
server.3=slave2:2888:3888


#编辑/etc/hosts ()
vim /etc/hosts
127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6



172.16.20.114 master1
172.16.20.113 slave1
172.16.20.123 slave2

#myid文件，里面的内容为数字，用于标识主机，如果这个文件没有的话，zooKeep无法启动（3台机器分别为1，2,3）
echo "1" >/data/ota_soft/zookeeper/data/myid

#启动
/data/ota_soft/zookeeper/bin/zkServer.sh start
 
#查看集群状态（都启动以后在查看状态）
 /data/ota_soft/zookeeper/bin/zkServer.sh status
可以看到一个leader，两个follower

2.kafka集群部署

2.1kafka部署三台机器

#官网下载压缩包

#上传服务器
cd /opt	

tar zxf kafka_2.11-1.0.0.tgz -C /data/ota_soft
 
ln -sf /data/ota_soft/kafka_2.11-1.0.0 /data/ota_soft/kafka
 
mkdir –p /data/ota_soft/kafka/logs

#修改配置文件
cd /data/ota_soft/kafka
vi /data/ota_soft/kafka/config/server.properties
############################# Server Basics #############################
broker.id=1   #对应修改（3台机器分别为1,2，3）
delete.topic.enable=true
############################# Socket Server Settings #############################
#host.name=192.168.10.22 
#listeners=PLAINTEXT://:9092
#advertised.listeners=PLAINTEXT://192.168.10.22:9092
listeners=PLAINTEXT://192.168.100.128:8180  #修改对应IP
advertised.host.name=master1  #不同机器修改对应名称
advertised.port=8180
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL
:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
#socket.request.max.bytes=104857600
socket.request.max.bytes=1677721600
############################# Log Basics #############################
log.dirs=/data/ota_soft/kafka/logs
num.partitions=3
default.replication.factor=3
num.recovery.threads.per.data.dir=1
############################# Log Flush Policy #############################
#log.flush.interval.messages=10000
#log.flush.interval.ms=1000
############################# Log Retention Policy #############################
log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
############################# Zookeeper #############################
zookeeper.connect=master1:2181,slave1:2181,slave2:2181
zookeeper.connection.timeout.ms=6000
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=2
transaction.state.log.min.isr=2
group.initial.rebalance.delay.ms=0

#启动
/data/ota_soft/kafka/bin/kafka-server-start.sh -daemon /data/ota_soft/kafka/config/server.properties

3.zookeepe部署报错

3.1启动报错

#1.权限报错
[root@localhost conf]$ /data/ota_soft/zookeeper/bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /data/ota_soft/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... /data/ota_soft/zookeeper/bin/zkServer.sh: line 140: ./zookeeper.out: Permission denied
/data/ota_soft/zookeeper/bin/zkServer.sh: line 149: /tmp/zookeeper/zookeeper_server.pid: Permission denied
FAILED TO WRITE PID

#解决方法
将配置文件下的bin、conf、data、logs属主属组与启动用户相同



#2.启动正常，查看集群状态报错
[root@localhost conf]$ /data/ota_soft/zookeeper/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /data/ota_soft/zookeeper/bin/../conf/zoo.cfg
Error contacting service. It is probably not running.

#解决方法
zookeeper的zoo.cfg的配置文件里的server123的IP顺序要相同（顺序决定于myid）
server.1=IP1:2888:3888
server.2=IP2:2888:3888
server.3=IP3:2888:3888

4.kafka部署报错

4.1启动报错

tail -100f server.log 
[2021-07-13 13:40:25,240] FATAL [KafkaServer id=203] Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kafka.common.KafkaException: Socket server failed to bind to IP:9092: Cannot assign requested address.


#解决方法
kafka的server.properties的配置文件listeners=PLAINTEXT://192.168.100.128:8180  IP是本机IP

5.zookeeper常用命令

链接zookeeper客户端：sh zkCli.sh -server 127.0.0.1:2181 （zookeeper部署在本机）
查看zk支持的所有命令：help （前提链接进客户端）
查看指定路径：ls
查看已连接的broker：ls /brokers/ids
查看所有topic：ls /brokers/topic
创建一个节点：create /zk mydate (创建一个/zk节点，且其内容为 “mydata”)
显示节点信息：get /zk
设置节点内容：set /zk “anotherData”
删除一个节点：delete /zk

6.kafka常用命令

启动服务：bin/kafka-server-start.sh config/server.properties &
停止服务：./kafka-server-stop.sh
查看所有话题：./kafka-topics.sh --list --zookeeper localhost:9092
查看所有话题的详细信息：./kafka-topics.sh --zookeeper localhost:2181 --describe
列出指定话题的详细信息：./kafka-topics.sh --zookeeper localhost:2181 --describe --topic test
删除一个话题： ./kafka-topics.sh --zookeeper localhost:2181 --delete --topic test
创建test话题，两个分区，每个分区三个副本：./kafka-topics.sh --zookeeper localhost:2181 --create --topic test --replication-factor 3 --partitions 2
生产消息：./kafka-console-producer.sh --broker-list localhost:9092 --topic test
消费消息：
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning ( #加了–from-beginning 重头消费所有的消息)
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test ( #不加–from-beginning 从最新的一条消息开始消费)
查看某个topic对应的消息数量：./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test --time -1
显示所有消费者：./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
获取正在消费的topic（console-consumer-63307）的group的offset：./kafka-consumer-groups.sh --describe --group console-consumer-63307 --bootstrap-server localhost:9092

7.下线一个broker

背景：主动下线是指broker运行正常，因为机器需要运维（升级操作系统，添加磁盘等）而主动停止broker

分两种情况处理：

若所有的topic的replica >= 2
此时，直接停止一个broker，会自动触发leader election操作，不过目前leader election是逐个partition进行，等待所有partition完成leader election耗时较长，这样不可服务的时间就比较长。为了缩短不可服务时间窗口，可以主动触发停止broker操作，这样可以逐个partition转移，直到所有partition完成转移，再停止broker。

/data/ota_soft/kafka/bin/kafka-run-class.sh kafka.admin.ShutdownBroker --zookeeper 127.0.0.1:2181 --broker 292 --num.retries 3 --retry.interval.ms 60

然后shutdown broker
/data/ota_soft/kafka/bin/kafka-server-stop.sh

若存在topic的replica=1

当存在topic的副本数小于2，只能手工把当前broker上这些topic对应的partition转移到其他broker上。当此broker上剩余的topic的replica > 2时，参照上面的处理方法继续处理。

大数据最新文章

实现Kafka至少消费一次

亚马逊云科技：还在苦于ETL？Zero ETL的时代

初探MapReduce

【SpringBoot框架篇】32.基于注解+redis实现

Elasticsearch：如何减少 Elasticsearch 集

Go redis操作

Redis面试题