kafka2集群部署

37次阅读

共计 6227 个字符，预计需要花费 16 分钟才能阅读完成。

kakfa 集群部署

准备工作

  1. jre 安装

zookeeper 单机集群搭建

假设当前目录为 /home/zk. 配置 3 节点伪集群.

  1. 下载
wget http://mirror.bit.edu.cn/apache/zookeeper/stable/apache-zookeeper-3.5.5-bin.tar.gz

zxf apache-zookeeper-3.5.5-bin.tar.gz

mv apache-zookeeper-3.5.5-bin zookeeper
  1. 创建 zookeeper data 目录
mkdir /home/zk/tmp/node-0/data -p
mkdir /home/zk/tmp/node-0/datalog -p
mkdir /home/zk/tmp/node-1/data -p
mkdir /home/zk/tmp/node-1/datalog -p
mkdir /home/zk/tmp/node-2/data -p
mkdir /home/zk/tmp/node-2/datalog -p
  1. 修改配置文件
cp zookeeper/conf/zoo_sample.cfg zookeeper/conf/zoo-node-0.cfg

# 修改节点 0 配置
vi zookeeper/conf/zoo-node-0.cfg
dataDir=/home/zk/tmp/node-0/data
dataLogDir=/home/zk/tmp/node-0/datalog

# server.index = ip:port- 原子广播:port- 选举
server.0=127.0.0.1:2887:3887
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889

# 拷贝节点 0 配置
cp zookeeper/conf/zoo-node-0.cfg zookeeper/conf/zoo-node-1.cfg
cp zookeeper/conf/zoo-node-0.cfg zookeeper/conf/zoo-node-2.cfg

# 修改节点 1 配置
vi zookeeper/conf/zoo-node-1.cfg
dataDir=/home/zk/tmp/node-1/data
dataLogDir=/home/zk/tmp/node-1/datalog
clientPort=2182

# 修改节点 2 配置
vi zookeeper/conf/zoo-node-2.cfg
dataDir=/home/zk/tmp/node-2/data
dataLogDir=/home/zk/tmp/node-2/datalog
clientPort=2183

# 指定节点编号
echo 0 >> tmp/node-0/data/myid
echo 1 >> tmp/node-1/data/myid
echo 2 >> tmp/node-2/data/myid
  1. 启动
cd zookeeper
bin/zkServer.sh start conf/zoo-node-0.cfg
bin/zkServer.sh start conf/zoo-node-1.cfg
bin/zkServer.sh start conf/zoo-node-2.cfg

# 查看状态
bin/zkServer.sh status conf/zoo-node-0.cfg
bin/zkServer.sh status conf/zoo-node-1.cfg
bin/zkServer.sh status conf/zoo-node-2.cfg

kafka 单机集群搭建

假设当前目录为 /home/kfk

  1. 下载
wget http://mirror.bit.edu.cn/apache/kafka/2.3.0/kafka_2.12-2.3.0.tgz
tar zxf kafka_2.12-2.3.0.tgz
# 或者设置软链接
mv kafka_2.12-2.3.0 kafka
  1. 创建日志目录
cd kafka

mkdir tmp/kafka-logs-0 -p
mkdir tmp/kafka-logs-1 -p
mkdir tmp/kafka-logs-2 -p
  1. 修改配置文件
cp config/server.properties config/server-0.properties

vi config/server-0.properties
broker.id=0
listeners=PLAINTEXT://:9092
log.dirs=/home/kfk/kafka/tmp/kafka-logs-0
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
# end vi

cp config/server-0.properties config/server-1.properties
cp config/server-0.properties config/server-2.properties

vi config/server-1.properties
broker.id=1
listeners=PLAINTEXT://:9093
log.dirs=/home/kfk/kafka/tmp/kafka-logs-1
# end vi

vi config/server-2.properties
broker.id=2
listeners=PLAINTEXT://:9094
log.dirs=/home/kfk/kafka/tmp/kafka-logs-2
# end vi
  1. 启动
nohup bin/kafka-server-start.sh config/server-0.properties 1>/dev/null 2>&1 &
nohup bin/kafka-server-start.sh config/server-1.properties 1>/dev/null 2>&1 &
nohup bin/kafka-server-start.sh config/server-2.properties 1>/dev/null 2>&1 &

常用配置说明

  1. broker 配置

https://kafka.apache.org/docu…

https://blog.csdn.net/memoord…

必要的配置

  • broker.id
  • log.dirs
  • zookeeper.connect
############################# Server Basics #############################
## 服务节点 id，整数 >=-1，集群内唯一
broker.id=0

######################## Socket Server Settings #########################
## socket 服务器监听地址
listeners=PLAINTEXT://:9092

############################# Log Basics ################################
## 存储日志文件的目录列表，逗号分隔
log.dirs=/home/yukl/2_opensys/kafka/tmp/kafka-logs-0

## 每个 topic 的默认分区数
num.partitions=1

######################## Internal Topic Settings  #######################
## offset topic 的复制因子
offsets.topic.replication.factor=1
## 事务 topic 的复制因子
transaction.state.log.replication.factor=1
## 事务 topic 的 min.insync.replicas 配置
transaction.state.log.min.isr=1

######################### Log Flush Policy ############################
## 消息数达到目标刷新到磁盘
#log.flush.interval.messages=10000

## 时间间隔达到目标刷新到磁盘
#log.flush.interval.ms=1000

######################### Log Retention Policy #########################
## 日志文件被删除前的保存时间（单位小时）log.retention.hours=168

## 优先使用 minutes 配置，否则使用 ms，如果这两项都没设置，则使用 hours
#log.retention.minutes
#log.retention.ms

## 日志达到删除大小的阈值。每个 topic 下每个分区保存数据的最大文件大小；注意，这是每个分区的上限，因此这个数值乘以分区的个数就是每个 topic 保存的数据总量。同时注意：如果 log.retention.hours 和 log.retention.bytes 都设置了，则超过了任何一个限制都会造成删除一个段文件。注意，这项设置可以由每个 topic 设置时进行覆盖。- 1 为不限制。#log.retention.bytes=1073741824

## topic 分区的日志存放在某个目录下诸多文件中，这些文件将 partition 的日志切分成一段一段的，这就是段文件（segment file）；一个 topic 的一个分区对应的所有 segment 文件称为 log。这个设置控制着一个 segment 文件的最大的大小，如果超过了此大小，就会生成一个新的 segment 文件。此配置可以被覆盖。int
log.segment.bytes=1073741824

## 检查日志段文件的间隔时间，以确定是否文件属性是否到达删除要求。log.retention.check.interval.ms=300000

############################# Zookeeper #############################
## zookeeper 主机地址
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183

## 与 zookeeper 建立连接的超时时间(ms)
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

## 消息中的时间戳取值为创建时间或者日志追加时间
## 取值 CreateTime 或者 LogAppendTime
## 默认 CreateTime
log.message.timestamp.type=CreateTime

  1. producer 配置

    https://kafka.apache.org/docu…

    name desc type default valid values
    acks 生产者的请求反馈方式：0- 不等待，仅仅将消息发送至缓冲区即返回，此模式下 retries 配置无效；1- 等待 leader 写成功即返回；all 或者 -1 表示等待所有的 follower 写成功才返回。 string 1 [all, -1, 0, 1]
    bootstrap.servers 初始连接服务列表 (host:port,host:port)， 是 kafka 集群的一个子集，用于发现所有的 kafak 集群 list “” non-null string
    compression.type 消息压缩模式，支持 none, gzip, snappy, lz4, or zstd. string none
    retries 设置一个比零大的值，客户端如果发送失败则会重新发送。注意，这个重试功能和客户端在接到错误之后重新发送没什么不同。如果 max.in.flight.requests.per.connection 没有设置为 1，有可能改变消息发送的顺序，因为如果 2 个批次发送到一个分区中，并第一个失败了并重试，但是第二个成功了，那么第二个批次将超过第一个 int 2147483647 [0,…,2147483647]
    client.id 一个字符串，在请求时传递给服务器。用于追踪请求源 string “”
    connections.max.idle.ms 连接闲置时间 long 540000
    delivery.timeout.ms 发送消息上报成功或失败的最大时间. 注意，如果此时间在 retries 耗尽之前达到，则判定请求失败 int “”

  2. consumer 配置

    https://kafka.apache.org/docu…

    name desc type default valid values
    bootstrap.servers 同 producer
    group.id consumer group 唯一标识 string
    auto.offset.reset earliest: 当各分区下有已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，从头开始消费 <br/>latest : 当各分区下有已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，消费新产生的该分区下的数据 <br/>none :topic 各分区都存在已提交的 offset 时，从 offset 后开始消费；只要有一个分区不存在已提交的 offset，则抛出异常 string latest [earliest,<br/>latest,<br/>none]
    connections.max.idle.ms 同上
    client.id 同上
    enable.auto.commit 如果为 true，消费者的 offset 后台会定期自动提交 boolean true
    auto.commit.interval.ms 消费者自动提交 offset 的周期（ms） int 5000

正文完
发表至： Kafka
2019-10-08
 0
《从0到1学习Flink》—— Flink 读取 Kafka 数据批量写入到 MySQL
关于kafka:javaspring-cloud-stream集成kafka和rabbitmq收发消息示例
关于kafka:kafka之三HW与leader-epoch
Kafka源码系列之Broker的IO服务及业务处理