共计 16164 个字符,预计需要花费 41 分钟才能阅读完成。
配置 /etc/hosts
文件
vi /etc/hosts
# 添加
192.168.200.110 master
192.168.200.111 slave1
192.168.200.112 slave2
安装 Zookeeper 集群
下载 Zookeeper 安装包
下载地址:https://www.apache.org/dyn/cl…
创建对应的 ZK 数据和日志目录
# 创建 ZK 的数据目录,同时需要创建 myid 指定这个节点的 ID
mkdir -p /software/zookeeper/zkdata/
vi /software/zookeeper/zkdata/myid
# 创建 ZK 的日志目录
mkdir /software/zookeeper/zklogs/
修改 zoo.cfg 文件
先复制一份:
cp zoo_sample.cfg zoo.cfg
开始修改:
cat zoo.cfg
# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
#dataDir=/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1
# ZK 的数据目录,myid 也在里面
dataDir=/software/zookeeper/zkdata
# ZK 的日志目录
dataLogDir=/software/zookeeper/zklogs
# 集群模式的各个节点,如果是自身的话,需要配置为 0.0.0.0,而不是 master/IP 地址,不然可能会出现 zk 之间无法连接通信的情况。(如果您提供了公共 IP,则侦听器将无法连接到端口,您必须为当前节点指定 0.0.0.0)
server.1=0.0.0.0:2888:3888
server.2=slave1:2888:3888
server.3=slave2:2888:3888
启动 / 停止 ZK/ 查看状态 / 重启
zkServer.sh start
zkServer.sh stop
zkServer.sh status
zkServer.sh restart
ZK 的相关命令操作
# 连接 ZK
[root@master conf]# zkCli.sh -server 127.0.0.1:2181
# 显示根目录下文件
[zk: 127.0.0.1:2181(CONNECTED) 0] ls /
[admin, brokers, cluster, config, consumers, controller_epoch, isr_change_notification, latest_producer_id_block, log_dir_event_notification, zookeeper]
# 显示根目录下文件,并能看到更新次数等数据
[zk: 127.0.0.1:2181(CONNECTED) 1] ls2 /
'ls2' has been deprecated. Please use 'ls [-s] path' instead.
[cluster, controller_epoch, brokers, zookeeper, admin, isr_change_notification, consumers, log_dir_event_notification, latest_producer_id_block, config]
cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x400000003
cversion = 10
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 10
# 创建文件,并设置初始化内容
[zk: 127.0.0.1:2181(CONNECTED) 3] create /lzhpo-test1 "hello,lzhpo-test1~"
Created /lzhpo-test1
# 查看文件内容
[zk: 127.0.0.1:2181(CONNECTED) 6] get /lzhpo-test1
hello,lzhpo-test1~
# 修改文件内容
[zk: 127.0.0.1:2181(CONNECTED) 7] set /lzhpo-test1 "test1"
[zk: 127.0.0.1:2181(CONNECTED) 8] get /lzhpo-test1
test1
# 删除文件
[zk: 127.0.0.1:2181(CONNECTED) 9] delete /lzhpo-test1
[zk: 127.0.0.1:2181(CONNECTED) 10] ls /
[admin, brokers, cluster, config, consumers, controller_epoch, isr_change_notification, latest_producer_id_block, log_dir_event_notification, zookeeper]
剩下的,请看官方文档:https://zookeeper.apache.org/…
安装 Kafka 集群
下载 Kafka 安装包
下载地址:https://www.apache.org/dyn/cl…
开启 JMX 监控(可选)
方法 1
如果不想做 kafka 的监控的话,可以选择不开启 JMX 监控。
就像下图这样子的监控,如果是自己写了一个 Kafka 的监控平台的话,不开启 JMX 监控就无法获取 Kafka 的一些信息。
修改 KAFKA_HEAP_OPTS
为主机运行内存的一半;
vi /software/kafka/kafka_2.12-2.2.0/bin/kafka-server-start.sh
添加export JMX_PORT="9999"
,也就是添加 JMX 监控,用来监控 Kafka 集群。
if ["x$KAFKA_HEAP_OPTS" = "x"]; then
export KAFKA_HEAP_OPTS="-Xmx2G -Xms2G"
export JMX_PORT="9999"
fi
方法 2(废弃)
这种方法在创建 topic 的时候,会报错:
Error: Exception thrown by the agent : java.rmi.server.ExportException: Port already in use: 9988; nested exception is:
java.net.BindException: Address in use (Bind failed)
sun.management.AgentConfigurationError: java.rmi.server.ExportException: Port already in use: 9988; nested exception is:
# 修改 kafka-run-class.sh
vi /software/kafka/kafka_2.12-2.2.0/bin/kafka-run-class.sh
# 第一行增加(也就是开启 JMX 监控,制定 JMX 监控端口)
JMX_PORT=9988
# 某些服务器可能无法正确绑定 ip,这时候我们需要显示指定绑定的 host
# 在参数 KAFKA_JMX_OPTS,增加一个,也就是制定服务 IP
# 不同的主机的需要修改为对应的 IP 地址
-Djava.rmi.server.hostname=192.168.10.110
修改bin/kafka-server-start.sh
这是我优化的地方,优化 JVM。
vi /software/kafka/kafka_2.12-2.2.0/bin/kafka-server-start.sh
# 调整 KAFKA_HEAP_OPTS="-Xmx16G -Xms16G”的值
# 推荐配置:一般 HEAP SIZE 的大小不超过主机内存的 50%。
修改config/server.properties
vi /software/kafka/kafka_2.12-2.2.0/bin/server.properties
# 修改处 1:broker.id,集群中唯一的 ID,不同主机的不能相同。broker.id=0
# 修改处 2:对外网开放地址。不同主机需要修改为对应的地址
listeners=PLAINTEXT://192.168.200.110:9092
# 修改处 3:监听地址。不同主机需要修改为对应的地址
advertised.listeners=PLAINTEXT://192.168.200.110:9092
# 修改处 4:socket 最大请求,int 类型的,不能超过 int 类型的最大值
socket.request.max.bytes=2147483600
# 修改处 5:partition 的个数
num.partitions=6
# 修改处 6:是否启用 log 压缩,一般不用启用,启用的话可以提高性能
log.cleaner.enable=false
# 修改处 7:zookeeper 集群地址
zookeeper.connect=192.168.200.110:2181,192.168.200.111:2181,192.168.200.112:2181
# 修改处 8:是否允许删除
# 是否允许删除 topic
delete.topic.enable=true
# 修改处 9:消息存放的目录,这个目录可以配置为“,”逗号分割的表达式,上面的 num.io.threads 要大于这个目录的个数这个目录,如果配置多个目录,新创建的 topic 他把消息持久化的地方是,当前以逗号分割的目录中,那个分区数最少就放那一个
log.dirs=/software/kafka/kafka-logs
# 创建消息存放的目录
mkdir -p /software/kafka/kafka-logs/
Kafka 调优请看我的文章:https://www.lzhpo.com/article…
配置文件调优详细说明(说明示例)
# 当前机器在集群中的唯一标识,和 zookeeper 的 myid 性质一样 建议用自己主机的后三位 每台(主机)broker 不一致
broker.id
# 当前 kafka 对外提供服务的端口默认是 9092 生产者(producer)要以这个端口为准[kafka-0.1.x 之前]
#port
# 这个参数默认是关闭的,在 0.8.1 有个 bug,DNS 解析问题,失败率的问题。(填写本机地址即可)[kafka-0.1.x 之前]
#host.name
# Kafka 启动的端口[kafka-0.1.x 之前]
#advertised.port=9092
# kafka 的本机 IP 地址[kafka-0.1.x 之前]
#advertised.host.name=192.168.10.130
# 监听
listeners=PLAINTEXT://192.168.10.130:9092
# 对外开放。kafka-0.1.x 之后的就是这样子修改的了,启用了上面的开放端口和地址方式
advertised.listeners=PLAINTEXT://192.168.10.130:9092
# 这个是 borker 进行网络处理的线程数
#【优化:num.network.threads 主要处理网络 io,读写缓冲区数据,基本没有 io 等待,配置线程数量为 cpu 核数加 1。】num.network.threads=3
# 这个是 borker 进行 I / O 处理的线程数
#【优化:num.io.threads 主要进行磁盘 io 操作,高峰期可能有些 io 等待,因此配置需要大些。配置线程数量为 cpu 核数 2 倍,最大不超过 3 倍。】num.io.threads=8
# 消息存放的目录,这个目录可以配置为“,”逗号分割的表达式,上面的 num.io.threads 要大于这个目录的个数这个目录,如果配置多个目录,新创建的 topic 他把消息持久化的地方是,当前以逗号分割的目录中,那个分区数最少就放那一个
log.dirs=/software/kafka/kafka-logs
# 是否允许删除 topic
delete.topic.enable=true
# 发送缓冲区 buffer 大小,数据不是一下子就发送的,先回存储到缓冲区了到达一定的大小后在发送,能提高性能
socket.send.buffer.bytes=102400
# kafka 接收缓冲区大小,当数据到达一定大小后在序列化到磁盘
socket.receive.buffer.bytes=102400
# 这个参数是向 kafka 请求消息或者向 kafka 发送消息的请请求的最大数,这个值不能超过 java 的堆栈大小
#【优化:这个是 int 类型的取值,int 类型范围是 -2147483648~2147483647。不能超出,超出之后报错:org.apache.kafka.common.config.ConfigException: Invalid value 8589934592 for configuration socket.request.max.bytes: Not a number of type INT。如果很纠结的话,那就按我推荐设置为 2147483600】socket.request.max.bytes=104857600
# 默认的分区数,一个 topic 默认 1 个分区数
#【优化:默认 partition 数量 1,如果 topic 在创建时没有指定 partition 数量,默认使用此值。Partition 的数量选取也会直接影响到 Kafka 集群的吞吐性能,配置过小会影响消费性能,建议改为 6。】num.partitions=1
# 默认消息的最大持久化时间,168 小时,7 天
log.retention.hours=168
# 消息保存的最大值 5M
message.max.byte=5242880
# kafka 保存消息的副本数,如果一个副本失效了,另一个还可以继续提供服务
default.replication.factor=2
# 取消息的最大直接数
replica.fetch.max.bytes=5242880
# 这个参数是:因为 kafka 的消息是以追加的形式落地到文件,当超过这个值的时候,kafka 会新起一个文件
log.segment.bytes=1073741824
# 每隔 300000 毫秒去检查上面配置的 log 失效时间(log.retention.hours=168),到目录查看是否有过期的消息如果有,删除
log.retention.check.interval.ms=300000
# 是否启用 log 压缩,一般不用启用,启用的话可以提高性能
log.cleaner.enable=false
# 设置 zookeeper 的连接端口 消费的时候要以这个端口消费
zookeeper.connect=192.168.10.130:2181,192.168.10.128:2181,192.168.10.129:2181
# zookeeper 连接超时时间
zookeeper.connection.timeout.ms=6000
#【优化:为了大幅度提高 producer 写入吞吐量,需要定期批量写文件。一般无需改动,如果 topic 的数据量较小可以考虑减少 log.flush.interval.ms 和 log.flush.interval.messages 来强制刷写数据,减少可能由于缓存数据未写盘带来的不一致。推荐配置分别 message 10000,间隔 1s。】# 每当 producer 写入 10000 条消息时,刷数据到磁盘
log.flush.interval.messages=10000
# 每间隔 1 秒钟时间,刷数据到磁盘
log.flush.interval.ms=1000
默认的 config/server.properties
配置文件:
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# see kafka.server.KafkaConfig for additional details and defaults
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0
############################# Socket Server Settings #############################
# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = listener_name://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092
#listeners=PLAINTEXT://:9092
# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured. Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
#advertised.listeners=PLAINTEXT://your.host.name:9092
# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3
# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8
# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400
# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400
# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600
############################# Log Basics #############################
# A comma separated list of directories under which to store log files
log.dirs=/tmp/kafka-logs
# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1
# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1
############################# Internal Topic Settings #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended for to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
############################# Log Flush Policy #############################
# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
# 1. Durability: Unflushed data may be lost if you are not using replication.
# 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
# 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.
# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000
# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000
############################# Log Retention Policy #############################
# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.
# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168
# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000
############################# Zookeeper #############################
# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=localhost:2181
# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000
############################# Group Coordinator Settings #############################
# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
group.initial.rebalance.delay.ms=0
修改完,分发到其它机器
scp -r /software/kafka/ slave1:/software/kafka/
scp -r /software/kafka/ slave2:/software/kafka/
修改其它机器的参数
config/server.properties
需要修改的参数,修改为自己主机的实际地址:
broker.id、listeners、advertised.listeners
bin/kafka-run-class.sh
需要修改的:
-Djava.rmi.server.hostname
kafka 相关操作命令
# 启动 kafka
[root@master /]# kafka-server-start.sh -daemon /software/kafka/kafka_2.12-2.2.0/config/server.properties
# 停止 kafka
[root@master /]# kafka-server-stop.sh
# 创建 topic,设置分区数为 3,副本数为 6
[root@master kafka-logs]# kafka-topics.sh --create --zookeeper 192.168.200.110:2181,192.168.200.111:2181,192.168.200.112:2181 --replication-factor 3 --partitions 6 --topic lzhpo-topic-test06
Created topic lzhpo-topic-test06.
# 查看集群所有 topic
[root@master kafka-logs]# kafka-topics.sh --list --zookeeper 192.168.200.110:2181,192.168.200.111:2181,192.168.200.112:2181
__consumer_offsets
lzhpo-topic-test01
lzhpo-topic-test02
lzhpo-topic-test03
lzhpo-topic-test04
lzhpo-topic-test05
lzhpo-topic-test06
lzhpo-topic-test3
lzhpo-topic-test4
topic-test-01
# 在 slave2 节点启动消费者 consumer,监听 lzhpo-topic-test3 的 topic
[root@slave2 ~]# kafka-console-consumer.sh --bootstrap-server 192.168.200.112:9092 --topic lzhpo-topic-test05 --from-beginning
# 在 slave1 启动一个生产者发送消息
[root@slave1 ~]# kafka-console-producer.sh --broker-list 192.168.200.111:9092 --topic lzhpo-topic-test05
slave2 节点启动一个消费者:
slave1 节点启动一个生产者发送消息:
# 查看 topic 的信息
[root@master kafka-logs]# kafka-topics.sh --describe --zookeeper master:2181,slave1:2181,slave2:2181 --topic lzhpo-topic-test05
Topic:lzhpo-topic-test05 PartitionCount:6 ReplicationFactor:3 Configs:
Topic: lzhpo-topic-test05 Partition: 0 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
Topic: lzhpo-topic-test05 Partition: 1 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2
Topic: lzhpo-topic-test05 Partition: 2 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
Topic: lzhpo-topic-test05 Partition: 3 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
Topic: lzhpo-topic-test05 Partition: 4 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1
Topic: lzhpo-topic-test05 Partition: 5 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
kafka 调优
调优参数:
-
replication.factor:这是给 topic 设置的参数,也就是分区数,这个参数值必须大于 1,默认是 1,要求每个 partition 必须至少有 2 个副本。kafka 集群虽然是高可用的,但是该 topic 在有 broker 宕机时,可能发生无法使用的情况。
-
方法 1:建立 topic 的时候就设置。
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 6 --topic lzhpo01
查看修改情况:
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic lzhpo01
-
方法 2:动态的给已经创建的 topic 添加
replication-factor
a. 首先我们配置 topic 的副本,保存为 json 文件()
例如,我们想把 lzhpo01 的部分设置为 3,(我的 kafka 集群有 3 个 broker,id 分别为 0,1,2),json 文件名称为 increase-replication-factor.json:{"version":1, "partitions":[{"topic":"lzhpo01","partition":0,"replicas":[0,1,2]}, {"topic":"lzhpo01","partition":1,"replicas":[0,1,2]}, {"topic":"lzhpo01","partition":2,"replicas":[0,1,2]} ]}
b. 执行脚本:
bin/kafka-reassign-partitions.sh -zookeeper 127.0.0.1:2181 --reassignment-json-file increase-replication-factor.json --execute
c. 查看修改之后的 topic 副本因子:
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic lzhpo01
参考官方文档:https://kafka.apache.org/docu…
-
-
min.insync.replicas:这是给 Kafka 服务端设置的参数,这个参数值也必须大于 1,意思是要求一个 Leader 至少有一个 Follower 跟自己保持联系,也就是数据和 Leader 一直是保持同步的,这样子就确保了 Leader 挂掉了,至少还有一个 Follower,不至于丢失数据。
设置
min.insync.replicas
:bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 6 --min.insync.replicas 2 --topic lzhpo01
- acks=all:这是在生产者端设置的,这是要求每条数据必须是写入所有的 replica 之后,才能认为是写入成功了。
就举个例子,SpringBoot 集成 Kafka 的话,可以直接在
application.yml
中配置:
-
retries=MAX:这个是要求一旦写入失败,就无限充值,卡在这里。这个我没有研究过,略 ……
欢迎关注我
一个热爱技术、热爱篮球的程序猿。
我的博客:https://www.lzhpo.com
我的微信公众号:会打篮球的程序猿