Kafka伪集群安装
Kaka集群安装 一、Kafka下载地址
Kafka下载页面：
https://kafka.apache.org/downloads
选择Kafka 3.2.0 版本进行下载：
https://www.apache.org/dyn/closer.cgi?path=/kafka/3.2.0/kafka_2.12-3.2.0.tgz
二、解压和配置
- 执行解压命令
tar -zxvf kafka_2.12-3.2.0.tgz- 进入Kafka目录
cd kafka_2.12-3.2.0- 创建cluster目录，用于保存Kafka集群相关的文件
# 创建cluster目录
mkdir cluster
# 进入cluster目录
cd cluster- 创建log目录用于存放Kafka集群运行日志
mkdir log
cd log
mkdir server1
mkdir server2
mkdir server3
mkdir zookeeper- 创建Broker1的配置文件
# 复制一份server.properties，命名为server1.propertiescp ../config/server.properties ./server1.properties# 修改server.properties配置文件
vim server.propertiesserver.properties文件配置项说明：
代码语言：javascript复制# 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.
#
# This configuration file is intended for use in ZK-based mode, where Apache ZooKeeper is required.
# 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的全局唯一编号，不能重复，和zookeeper的myid是一个意思
broker.id=1############################# Socket Server Settings #############################
# The address the socket server listens on. If not configured, the host name will be equal to the value of# java.net.InetAddress.getCanonicalHostName(), with PLAINTEXT listener name, and port 9092.# FORMAT:
# listeners = listener_name://host_name:port# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092# broker监听IP和端口也可以是域名
listeners=PLAINTEXT://:9091# Listener name, hostname and port the broker will advertise to clients.
# If not set, it uses the value for "listeners".
#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
# 用于处理请求的线程数量，包括磁盘IO请求，这个数量和log.dirs配置的目录数量有关，这里的数量不能小于log.dirs的数量，
# 虽然log.dirs是配置日志存放路径，但是它可以配置多个目录后面用逗号分隔
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)
# 向kafka套接字请求最大字节数量，防止服务器OOM，也就是OutOfMemery，这个数量不要超过JAVA的堆栈大小，
socket.request.max.bytes=104857600############################# Log Basics #############################
# A comma separated list of directories under which to store log files
# 日志路径也就是分区日志存放的地方，你所建立的topic的分区就在这里面，但是它可以配置多个目录后面用逗号分隔
log.dirs=./log/server1
# 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.
# 默认分区数量，当建立Topic时不指定分区数量，默认就1num.partitions=3# 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.
# segment文件默认会被保留7天的时间，超时的话就会被清理，那么清理这件事情就需要有一些线程来做。
# 这里就是用来设置恢复和清理data下数据的线程数量
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 to ensure availability such as 3.# kafka的内部topic consumer_offsets副本数。
# 如果consumer_offsets副本数设置为1，当该副本所在的broker宕机，consumer_offsets只有一份副本，该分区宕机。
# 使用该分区存储消费分组offset位置的消费者均会收到影响，offset无法提交，从而导致生产者可以发送消息但消费者不可用。所以需要设置该字段的值大于1。
offsets.topic.replication.factor=3# 事务主题的复制因子（设置更高以确保可用性）。
transaction.state.log.replication.factor=1# 覆盖事务主题的min.insync.replicas配置
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文件"sync"到磁盘之前累积的消息条数，因为磁盘IO操作是一个慢操作,但又是一个"数据可靠性"的必要手段
# 所以此参数的设置,需要在"数据可靠性"与"性能"之间做必要的权衡.
# 如果此值过大,将会导致每次"fsync"的时间较长(IO阻塞)
# 如果此值过小,将会导致"fsync"的次数较多,这也意味着整体的client请求有一定的延迟.
# 物理server故障,将会导致没有fsync的消息丢失.
#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.cleanup.policy设置的策略处理数据，也就是消费端能够多久去消费数据
# log.retention.bytes和log.retention.hours任意一个达到要求，都会执行删除
# 如果你创建Topic的时候指定了这个参数，那么你以你指定的为准
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.
# 日志文件中每个segment的大小，默认为1G。topic的分区是以一堆segment文件存储的，这个控制每个segment的大小，当超过这个大小会建立一个新日志文件
# 这个参数会被topic创建时的指定参数覆盖，如果你创建Topic的时候指定了这个参数，那么你以你指定的为准。
log.segment.bytes=1073741824# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
# 上面的参数设置了每一个segment文件的大小是1G，那么就需要有一个东西去定期检查segment文件有没有达到1G，多长时间去检查一次，
# 就需要设置一个周期性检查文件大小的时间（单位是毫秒）。
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集群地址
zookeeper.connect=localhost:2181# Timeout in ms for connecting to zookeeper
#配置连接Zookeeper集群的超时时间
zookeeper.connection.timeout.ms=18000############################# 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- 创建Broker2、Broker3的配置文件
【注】因为是在本地构建的Kafka伪集群，broker.id和listeners需要确保不能重复。
- 复制Zookeeper配置文件
cp ../cluster/zookeeper.properties ./zookeeper.properties- 创建启动脚本
vim start.sh# 启动Zookeepernohup ../bin/zookeeper-server-start.sh ./zookeeper.properties &
# 启动Broker1../bin/kafka-server-start.sh -daemon ./server1.properties# 启动Broker2../bin/kafka-server-start.sh -daemon ./server2.properties# 启动Broker3../bin/kafka-server-start.sh -daemon ./server3.propertieswq- 启动Kafka集群
sh start.sh- 验证启动结果
# 创建topic名称为test
../bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092# 执行结果Created topic test.- 创建关闭脚本
vim stop.sh../bin/kafka-server-stop.sh stop ./server1.properties../bin/kafka-server-stop.sh stop ./server2.properties../bin/kafka-server-stop.sh stop ./server3.properties../bin/zookeeper-server-stop.sh ./zookeeper.propertieswq- 执行Kafka集群
sh stop.sh
