Kafka部署篇

时间 2019-12-05

标签 kafka 部署栏目 Kafka 繁體版

原文原文链接

目录html

安装

下载与安装

kafka下载地址：https://kafka.apache.org/downloads
须要说明的是，kafka的安装依赖于zk，zk的部署可直接参考《Zookeeper介绍与基本部署》。固然，kafka默认也内置了zk的启动脚本，在kafka安装路径的bin目录下，名称为zookeeper-server-start.sh，若是不想独立安装zk，可直接使用该脚本。java

wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz
tar xf kafka_2.12-2.2.0.tgz -C /usr/local/
cd /usr/local
ln -s kafka_2.12-2.2.0 kafka

配置

kafka主配置文件为/usr/local/kafka/config/server.properties，配置示例以下：apache

broker.id=0
listeners=PLAINTEXT://10.1.60.29:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/logs
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.1.60.29:2181,10.1.61.195:2181,10.1.61.27:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
auto.create.topics.enable=true
delete.topics.enable=true

配置说明：json

broker.id：每一个broker在集群中的惟一标识，正整数。当该服务器的ip地址发生变动，但broker.id未变，则不会影响consumers的消费状况
listeners：kafka的监听地址与端口，在实际测试中若是写0.0.0.0会报错。
num.network.threads：kafka用于处理网络请求的线程数
num.io.threads：kafka用于处理磁盘io的线程数
socket.send.buffer.bytes：发送数据的缓冲区
socket.receive.buffer.bytes：接收数据的缓冲区
socket.request.max.bytes：容许接收的最大数据包的大小（防止数据包过大致使OOM）
log.dirs：kakfa用于保存数据的目录，全部的消息都会存储在该目录当中。能够经过逗号来指定多个路径，kafka会根据最少被使用的原则选择目录分配新的partition。须要说明的是，kafka在分配partition的时候选择的原则不是按照磁盘空间大小来定的，而是根据分配的partition的个数多少而定
num.partitions：设置新建立的topic的默认分区数
number.recovery.threads.per.data.dir：用于恢复每一个数据目录时启动的线程数
log.retention.hours：配置kafka中消息保存的时间，还支持log.retention.minutes和log.retention.ms。若是多个同时设置会选择时间最短的配置，默认为7天。
log.retention.check.interval.ms：用于检测数据过时的周期
log.segment.bytes：配置partition中每一个segment数据文件的大小。默认为1GB。超出该大小后，会自动建立一个新的segment文件。
zookeeper.connect：指定链接的zk的地址，zk中存储了broker的元数据信息。能够经过逗号来设置多个值。格式为：hostname:port/path。hostname为zk的主机名或ip，port为zk监听的端口。/path表示kafka的元数据存储到zk上的目录，若是不设置，默认为根目录
zookeeper.connection.timeout：kafka链接zk的超时时间
group.initial.rebalance.delay.ms：在实际环境当中，当将多个consumer加入到一个空的consumer group中时，每加入一个consumer就会触发一次对partition消费的重平衡，若是加入100个，就得重平衡100次，这个过程就会变得很是耗时。经过设置该参数，能够延迟重平衡的时间，好比有100个consumer会在10s内所有加入到一个consumer group中，就能够将该值设置为10s，10s以后，只须要作一次重平衡便可。默认为0则表明不开启该特性。
auto.create.topics.enable：当有producer向一个不存在的topic中写入消息时，是否自动建立该topic
delete.topics.enable：kafka提供了删除topic的功能，但默认并不会直接将topic数据物理删除。若是要从物理上删除（删除topic后，数据文件也一并删除），则须要将此项设置为true

须要说明的是，多个kafka节点依赖zk实现集群，因此各节点并不须要做特殊配置，只须要broker.id不一样，并接入到同一个zk集群便可。bootstrap

启停操做

#启动
/usr/local/kafka/bin/kafka-server-start.sh  -daemon /usr/local/kafka/config/server.properties 

#检查java进程

# jps
1394 QuorumPeerMain
13586 Logstash
27591 Kafka
27693 Jps

#中止
/usr/local/kafka/bin/kafka-server-start.sh

验证

能够经过zookeeper查看kafka的元数据信息：服务器

#经过zk客户端链接zookeeper

../zookeeper/bin/zkCli.sh 

#查看根下多了不少目录
[zk: localhost:2181(CONNECTED) 1] ls /
[cluster, controller_epoch, controller, brokers, zookeeper, admin, isr_change_notification, consumers, log_dir_event_notification, latest_producer_id_block, config]

#查看/brokers/ids，能够看到有三个broker已经加入
[zk: localhost:2181(CONNECTED) 8] ls /brokers/ids
[0, 1, 2]

#查看/brokers/topics，目前为空，说明尚未建立任何的topic
[zk: localhost:2181(CONNECTED) 3] ls /brokers/topics
[]

基本操做

建立topic

上面完成了kafka的部署，经过验证部署咱们发现当前没有topic，因此建立一个topic以下：网络

# ./bin/kafka-topics.sh --create --zookeeper localhost:2181  --replication-factor 2 --partitions 3 --topic myfirsttopic 
Created topic myfirsttopic.

参数说明：socket

--create：建立一个topic
--zookeeper: 指定zookeeper集群的主机列表，多个server可以使用逗号分隔，这里由于kafka和zk是在同一个server，因此直接链接了本机的2181端口
--replication-factor：指定建立这个topic的副本数
--partitions：指定该topic的分区数
--topic：指定topic的名称

列出现有的topic

上面经过操做zk就能够看到topic相关信息，接下来咱们直接经过kafka命令行来进行相关操做：测试

# ./bin/kafka-topics.sh --zookeeper localhost:2181 --list                                                               
myfirsttopic

查看topic的详细信息

#查看myfirsttopic的详细信息
# ./bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic myfirsttopic
Topic:myfirsttopic      PartitionCount:3        ReplicationFactor:2     Configs:
        Topic: myfirsttopic     Partition: 0    Leader: 0       Replicas: 0,2   Isr: 0
        Topic: myfirsttopic     Partition: 1    Leader: 1       Replicas: 1,0   Isr: 1,0
        Topic: myfirsttopic     Partition: 2    Leader: 2       Replicas: 2,1   Isr: 2,1

参数说明：大数据

--describe：查看topic的详细信息

输出说明：

leader：当前负责读写的leader broker
replicas：当前分区全部的副本对应的broker列表
isr：处于活动状态的broker

增长topic的partition数量

# ./bin/kafka-topics.sh --zookeeper localhost:2181 --alter --partitions 6 --topic myfirsttopic
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!

# ./bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic myfirsttopic            
Topic:myfirsttopic      PartitionCount:6        ReplicationFactor:2     Configs:
        Topic: myfirsttopic     Partition: 0    Leader: 0       Replicas: 0,2   Isr: 0
        Topic: myfirsttopic     Partition: 1    Leader: 1       Replicas: 1,0   Isr: 1,0
        Topic: myfirsttopic     Partition: 2    Leader: 2       Replicas: 2,1   Isr: 2,1
        Topic: myfirsttopic     Partition: 3    Leader: 0       Replicas: 0,2   Isr: 0,2
        Topic: myfirsttopic     Partition: 4    Leader: 1       Replicas: 1,0   Isr: 1,0
        Topic: myfirsttopic     Partition: 5    Leader: 2       Replicas: 2,1   Isr: 2,1

修改一个topic的副本数

#建立一个topic名为mysecondtopic，指定分区为2，副本为1
# ./bin/kafka-topics.sh --zookeeper localhost:2181 --create --replication-factor 1 --partitions 2 --topic mysecondtopic
Created topic mysecondtopic.

#查看新建立的topic详细信息
# ./bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic mysecondtopic
Topic:mysecondtopic     PartitionCount:2        ReplicationFactor:1     Configs:
        Topic: mysecondtopic    Partition: 0    Leader: 0       Replicas: 0     Isr: 0
        Topic: mysecondtopic    Partition: 1    Leader: 1       Replicas: 1     Isr: 1

#将broker.id为0上的partition的副本由原来的[0]扩充为[0,2]，将broker.id为1上的partition的副本由原来的[1]扩充为[1,2]。
#须要先建立一个json文件以下：
# cat partitions-to-move.json
{
    "partitions":
    [
        {
            "topic":"mysecondtopic",
            "partition": 0,
            "replicas": [0,2]
        },
        {
            "topic": "mysecondtopic",
            "partition": 1,
            "replicas": [1,2]
        }
    ],
    "version": 1
}

#执行副本修改
# ./bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ./partitions-to-move.json --execute
Current partition replica assignment

{"version":1,"partitions":[{"topic":"mysecondtopic","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"mysecondtopic","partition":0,"replicas":[0],"log_dirs":["any"]}]}

Save this to use as the --reassignment-json-file option during rollback
Successfully started reassignment of partitions.

#再次查看topic状态，发现副本数由按照预期发生变动
# ./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic mysecondtopic
Topic:mysecondtopic     PartitionCount:2        ReplicationFactor:2     Configs:
        Topic: mysecondtopic    Partition: 0    Leader: 0       Replicas: 0,2   Isr: 0
        Topic: mysecondtopic    Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1

删除一个topic

#执行删除操做
# ./bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic myfirsttopic
Topic myfirsttopic is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

#查看topic，能够看到myfirsttopic已被删除
# ./bin/kafka-topics.sh --zookeeper localhost:2181 --list
__consumer_offsets
mysecondtopic

经过producer生产消息

# ./bin/kafka-console-producer.sh  --broker-list 10.1.60.29:9092 --topic mysecondtopic                                                
>hello kafka!
>hello world!
>just a test!
>
>hi world!
>hahahaha!
>

经过consumer消费消息

# ./bin/kafka-console-consumer.sh --bootstrap-server 10.1.60.29:9092 --topic mysecondtopic --from-beginning       
hello kafka!
just a test!
hi world!
hello world!

hahahaha!