centos7 安装kafka

centos7 安装kafka_2.12-2.5.0

访问量:1396 创建时间:2020-05-14

在使用过程中，发现安装kafka_2.12-2.7 时java客户端无法连接kafka, 更换kafka_2.13-2.7 后，java客户端可以连接kafka。

介绍

三个主要功能

消息订阅发布
以容错的持久方式存储数据
数据流实时处理

概念

Kafka作为一个集群在一个或多个服务器上运行，这些服务器可以跨越多个数据中心。
Kafka集群将记录流存储在topics中。
每个记录由一个键、一个值和一个时间戳组成。

5个核心api

Producer API :允许应用程序将记录流发布到一个或多个Kafka主题。
Consumer API:允许应用程序订阅一个或多个主题并处理产生给它们的记录 Streams API : 允许应用程序充当流处理器，使用来自一个或多个主题的输入流，并将输出流生成到一个或多个输出主题，从而有效地将输入流转换为输出流。
Streams API : 允许应用程序充当流处理器，使用来自一个或多个主题的输入流，并将输出流生成到一个或多个输出主题，从而有效地将输入流转换为输出流。
Connector API: 允许构建和运行将Kafka主题连接到现有应用程序或数据系统的可重用生产者或消费者。例如，到关系数据库的连接器可能会捕获对表的每个更改。
Admin API : 允许管理和检查主题、代理和其他Kafka对象。

topic与 logs

对于每个主题，Kafka集群维护一个分区日志，如下所示:

每个分区都是一个有序的、不可变的记录序列，不断地追加到一个结构化提交日志中。分区中的每个记录都分配了一个连续的id号，称为偏移量，它惟一地标识分区中的每个记录。
Kafka的性能在数据大小方面是稳定的（不会因为数据很大性能变差），所以长时间存储数据不是问题。
Kafka集群使用一个可配置的保留期（例如10天）持久保存所有已发布的记录，超过时间的数据会被删除。

数据复制与容错

分区分布在Kafka群集中的服务器上，每个服务器处理数据并要求共享分区。每个分区都在可配置数量的服务器之间复制，以实现容错功能。每个分区有一个leader，有0-多个followers，leader处理所有的读写请求，followers被动的复制leader数据，leader故障后，follower按照规则升级为leader。

消费者

消费者按组划分，相同组的消费者会平均消费topic的分区（同组的多个消费者不能消费一个分区）
一个topic可以被多个消费者组消费，每个组获取的数据都一样。
消费者组中的consumer实例数不能超过分区数。

kafka_2.12-2.5.0 安装

kafka清华大学下载地址 zookeeper下载地址

[root@localhost ~]# wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.5.0/kafka_2.12-2.5.0.tgz
[root@k8s-n1 ~]# wget https://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.6.1/apache-zookeeper-3.6.1-bin.tar.gz

zookeeper安装请移步

机器信息:

IP	hostname&app
192.168.0.126	k8s-n2& zookeeper、kafka
192.168.0.127	k8s-n3& zookeeper、kafka
192.168.0.128	k8s-n4& zookeeper、kafka

[root@k8s-n2 ~]# tar xf kafka_2.12-2.5.0.tgz 
[root@k8s-n2 ~]# mv kafka_2.12-2.5.0 /data/kafka
[root@k8s-n2 ~]# cd /data/kafka
[root@k8s-n2 kafka]# mkdir -pv /data/kafka/logs
#修改配置文件（主要修改log.dirs、zookeeper.connect；broker.id三台机器不能相同）
[root@k8s-n2 kafka]# cat config/server.properties | egrep -v '^#|^$'
broker.id=0
#broker.id=1 #192.168.0.127机器id设置
#broker.id=2 #192.168.0.128机器id设置
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=1
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=192.168.0.126:2181,192.168.0.127:2181,192.168.0.128:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
#启动（3节点都启动）
[root@k8s-n2 kafka]# ./bin/kafka-server-start.sh -daemon config/server.properties 
[root@k8s-n2 kafka]# jps
1361 QuorumPeerMain
28581 Jps
27815 Kafka
#关闭命令./bin/kafka-server-stop.sh

至此centos7 kafka安装过程结束

简单使用

#创建topic
[root@k8s-n2 kafka]# ./bin/kafka-topics.sh --create --bootstrap-server 192.168.0.126:9092 --replication-factor 3 --partitions 1 --topic my-replicated-topic
Created topic my-replicated-topic.
#查看topci
[root@k8s-n2 kafka]# ./bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic
Topic: my-replicated-topic  PartitionCount: 1   ReplicationFactor: 3    Configs: segmen
    Topic: my-replicated-topic  Partition: 0    Leader: 2   Replicas: 2,0,1 Isr: 2,
[root@k8s-n2 kafka]# 
#启动一个生产者，发送消息
[root@k8s-n2 kafka]# bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic my-replicated-topic
>my test message 1
>my test message 2   
>^C
#启动一个消费者消费消息
[root@k8s-n2 kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-inning --topic my-replicated-topic
my test message 1
my test message 2
^C
#杀死一个kafka
[root@k8s-n2 kafka]# jps
1361 QuorumPeerMain
17109 Jps
30809 Kafka
[root@k8s-n2 kafka]# kill 30809
[root@k8s-n2 kafka]# jps
1361 QuorumPeerMain
17401 Jps
#在另外一台机器查看,Isr同步的只有2个节点
[root@k8s-n3 kafka]# bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic my-replicated-topic
Topic: my-replicated-topic  PartitionCount: 1   ReplicationFactor: 3    Configs: segment.bytes=1073741824
    Topic: my-replicated-topic  Partition: 0    Leader: 2   Replicas: 2,0,1 Isr: 2,1
#再发送消息
[root@k8s-n2 kafka]#  bin/kafka-console-producer.sh --bootstrap-server 192.168.0.127:9092 --topic my-replicated-topic
>111
>222
>^C
[root@k8s-n2 kafka]#  bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.127:9092 --from-beginning --topic my-replicated-topic 
my test message 1
my test message 2
111
222
^C
#启动kafka，再次查看Isr状态，很快恢复
[root@k8s-n2 kafka]# ./bin/kafka-server-start.sh -daemon config/server.properties 
[root@k8s-n2 kafka]# ./bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic my-replicated-topic
Topic: my-replicated-topic  PartitionCount: 1   ReplicationFactor: 3    Configs: segment.bytes=1073741824
    Topic: my-replicated-topic  Partition: 0    Leader: 2   Replicas: 2,0,1 Isr: 2,1,0

kafka 2.7 单机安装

下载地址

jdk 支持8和11版本，这里使用jdk8，jdk安装过程省略

[root@localhost data]# tar xf kafka_2.12-2.7.0.tgz
[root@localhost data]# cd kafka_2.12-2.7.0/
[root@localhost kafka_2.12-2.7.0]# ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties
##关闭zk命令使用 ./bin/zookeeper-server-stop.sh -daemon ./config/zookeeper.properties
#启动kafka
[root@localhost kafka_2.12-2.7.0]# ./bin/kafka-server-start.sh -daemon ./config/server.properties
###关闭kafka命令./bin/kafka-server-stop.sh ./config/server.properties

<< 上一章节

下一章节 >>

登陆评论: 使用GITHUB登陆