当前位置：首页 > article >正文

部署kafka

article 2024/11/19 22:49:29

kafka简介：

Kafka是一种分布式流处理平台，也是一种高吞吐量、低延迟的分布式发布订阅消息系统，由Apache软件基金会开发。它被设计成可扩展的、高容错的，具有高吞吐量的、低延迟的数据处理管道。Kafka可以应用于许多不同的场景，例如实时流数据处理、日志收集、报警和监控等，可提供可靠、高效、集中化的数据传输。Kafka具有以下优点：

高吞吐量：Kafka可以处理大量数据的读写请求，支持每秒几百万条消息的处理。
可扩展性：Kafka是一个分布式系统，可以水平扩展，支持集群配置，增加节点可以提高系统吞吐量。
持久性：Kafka可以将数据持久化到磁盘上，支持高可靠性和持久性存储。
实时性：Kafka支持实时数据处理，能够在毫秒级别内处理消息。
多语言支持：Kafka提供多种语言的客户端API，支持Java、Python、Go、C++等多种编程语言。

总之，Kafka是一个高性能、可靠、分布式的消息中间件，广泛应用于日志收集、消息队列、数据处理、实时计算等领域。

部署kafka

服务器

安装软件	主机名	IP地址	系统版本	配置
zookeeper/kafka	kafka-1	192.168.231.249	centos7.5.1804	1核2G
zookeeper/kafka	kafka-2	192.168.231.250	centos7.5.1804	1核2G
zookeeper/kafka	kafka-3	192.168.231.251	centos7.5.1804	1核2G

软件版本：jdk-8u121-linux-x64.tar.gz、kafka_2.11-2.0.0.tgz
示例节点： 192.168.231.249

安装配置jdk8

三台机器都得安装

Kafka、Zookeeper（简称：ZK）运行依赖jdk8

解压
[root@kafka-1 ~]# tar zxvf /usr/local/package/jdk-8u121-linux-x64.tar.gz -C /usr/local/

配置环境变量
[root@kafka-1 ~]# cat /etc/profile.d/java.sh 
JAVA_HOME=/usr/local/java
PATH=$PATH:$JAVA_HOME/bin

环境变量生效
[root@kafka-1 ~]# source /etc/profile

检查是否安装成功
[root@kafka-1 ~]# java -version
java version "1.8.0_211"
Java(TM) SE Runtime Environment (build 1.8.0_211-b12)
Java HotSpot(TM) 64-Bit Server VM (build 25.211-b12, mixed mode)

安装配置ZK

Kafka运行依赖ZK，Kafka官网提供的tar包中，已经包含了ZK，这里不再额下载ZK程序。

安装

三台机器都安装

[root@kafka-1 ~]# tar zxvf /usr/local/package/kafka_2.11-2.0.0.tgz -C /usr/local/

配置

[root@kafka-1 ~]# echo '
dataDir=/opt/data/zookeeper/data 
dataLogDir=/opt/data/zookeeper/logs
clientPort=2181 
tickTime=2000 
initLimit=20 
syncLimit=10 
server.1=192.168.231.249:2888:3888             //kafka集群IP:Port .1为id 3处要对应
server.2=192.168.231.250:2888:3888
server.3=192.168.231.251:2888:3888
'> /usr/local/kafka_2.11-2.0.0/config/zookeeper.properties

配置项含义

dataDir ZK数据存放目录。
dataLogDir  ZK日志存放目录。
clientPort  客户端连接ZK服务的端口。
tickTime        ZK服务器之间或客户端与服务器之间维持心跳的时间间隔。
initLimit       允许follower(相对于Leaderer言的“客户端”)连接并同步到Leader的初始化连接时间，以tickTime为单位。当初始化连接时间超过该值，则表示连接失败。
syncLimit   Leader与Follower之间发送消息时，请求和应答时间长度。如果follower在设置时间内不能与leader通信，那么此follower将会被丢弃。
server.1=192.168.231.249:2888:3888    2888是follower与leader交换信息的端口，3888是当leader挂了时用来执行选举时服务器相互通信的端口。

创建数据日志目录

三台机器都得创建

[root@kafka-1 ~]# mkdir -p /opt/data/zookeeper/{data,logs}
创建myid文件
[root@kafka-1 ~]# echo 1 > /opt/data/zookeeper/data/myid

kafka-2
[root@kafka-2 ~]# mkdir -p /opt/data/zookeeper/{data,logs}
创建myid文件
[root@kafka-2 ~]# echo 2 > /opt/data/zookeeper/data/myid

kafka-3
[root@kafka-3 ~]# mkdir -p /opt/data/zookeeper/{data,logs}
创建myid文件
[root@kafka-3 ~]# echo 3 > /opt/data/zookeeper/data/myid

配置kafka

配置

kafka-1
[root@kafka-1 ~]# echo '
broker.id=1
listeners=PLAINTEXT://192.168.231.249: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=/opt/data/kafka/logs
num.partitions=6
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=536870912
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.231.149:2181,192.168.231.250:2181,192.168.231.251:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
' >/usr/local/kafka_2.11-2.0.0/config/server.properties

kafka-2
[root@kafka-2 ~]# echo '
broker.id=2
listeners=PLAINTEXT://192.168.231.250: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=/opt/data/kafka/logs
num.partitions=6
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=536870912
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.231.249:2181,192.168.231.250:2181,192.168.231.251:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
' >/usr/local/kafka_2.11-2.0.0/config/server.properties

kafka-3
[root@kafka-1 ~]# echo '
broker.id=3
listeners=PLAINTEXT://192.168.231.251: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=/opt/data/kafka/logs
num.partitions=6
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=536870912
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.231.249:2181,192.168.231.250:2181,192.168.231.251:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0' >/usr/local/kafka_2.11-2.0.0/config/server.properties

配置项含义

broker.id   每个server需要单独配置broker id，如果不配置系统会自动配置。
listeners       监听地址，格式PLAINTEXT://IP:端口。
num.network.threads 接收和发送网络信息的线程数。
num.io.threads          服务器用于处理请求的线程数，其中可能包括磁盘I/O。
socket.send.buffer.bytes    套接字服务器使用的发送缓冲区(SO_SNDBUF)
socket.receive.buffer.bytes 套接字服务器使用的接收缓冲区(SO_RCVBUF)
socket.request.max.bytes        套接字服务器将接受的请求的最大大小(防止OOM)
log.dirs        日志文件目录。
num.partitions  partition数量。
num.recovery.threads.per.data.dir       在启动时恢复日志、关闭时刷盘日志每个数据目录的线程的数量，默认1。
offsets.topic.replication.factor        偏移量话题的复制因子（设置更高保证可用），为了保证有效的复制，偏移话题的复制因子是可配置的，在偏移话题的第一次请求的时候可用的broker的数量至少为复制因子的大小，否则要么话题创建失败，要么复制因子取可用broker的数量和配置复制因子的最小值。
log.retention.hours 日志文件删除之前保留的时间（单位小时），默认168
log.segment.bytes   单个日志文件的大小，默认1073741824
log.retention.check.interval.ms 检查日志段以查看是否可以根据保留策略删除它们的时间间隔。
zookeeper.connect   ZK主机地址，如果zookeeper是集群则以逗号隔开。
zookeeper.connection.timeout.ms     连接到Zookeeper的超时时间

创建log目录

[root@kafka-1 ~]# mkdir -p /opt/data/kafka/logs

[root@kafka-2 ~]# mkdir -p /opt/data/kafka/logs

[root@kafka-3 ~]# mkdir -p /opt/data/kafka/logs

启动，验证ZK集群

启动

三个节点依次执行

[root@kafka-1 ~]# cd /usr/local/kafka_2.11-2.0.0/

放入后台运行
[root@kafka-1 ~]# nohup bin/zookeeper-server-start.sh config/zookeeper.properties &

验证，查看ZK配置

下载nmap或者直接下载nc
[root@kafka-1 ~]# yum install nmap  或者yum -y install nc
[root@kafka-1 ~]# echo conf | nc 127.0.0.1 2181
clientPort=2181
dataDir=/opt/data/zookeeper/data/version-2
dataLogDir=/opt/data/zookeeper/logs/version-2
tickTime=2000
maxClientCnxns=0
minSessionTimeout=4000
maxSessionTimeout=40000
serverId=1
initLimit=20
syncLimit=10
electionAlg=3
electionPort=3888
quorumPort=2888
peerType=0

查看ZK状态

[root@kafka-1 ~]# echo stat | nc 127.0.0.1 2181
Zookeeper version: 3.4.13-2d71af4dbe22557fda74f9a9b4309b15a7487f03, built on 06/29/2018 00:39 GMT
Clients:
 /127.0.0.1:35088[0](queued=0,recved=1,sent=0)

Latency min/avg/max: 0/3/21
Received: 19
Sent: 18
Connections: 1
Outstanding: 0
Zxid: 0x3000000b6
Mode: follower
Node count: 135

查看端口

[root@kafka-1 ~]# netstat -nplt | grep 2181
tcp6       0      0 :::2181                 :::*                    LISTEN      1420/java

启动验证kafka

启动

在三个节点依次执行

[root@kafka-1 ~]# cd /usr/local/kafka_2.11-2.0.0/
后台执行
[root@kafka-1 ~]# nohup bin/kafka-server-start.sh config/server.properties &

kafka-2
[root@kafka-2 ~]# cd /usr/local/kafka_2.11-2.0.0/
[root@kafka-2 ~]# nohup bin/kafka-server-start.sh config/server.properties &

kafka-3
[root@kafka-3 ~]# cd /usr/local/kafka_2.11-2.0.0/
[root@kafka-3 ~]# nohup bin/kafka-server-start.sh config/server.properties &

验证

在192.168.231.249创建topic

[root@kafka-1 ~]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testtopic

屏幕回显
Created topic "testtopic".

查询192.168.231.249上的topic

[root@kafka-1 ~]# bin/kafka-topics.sh --zookeeper 192.168.231.249:2181 --list               

屏幕回显
testtopic

查询192.168.231.250上的topic

[root@kafka-1 kafka_2.11-2.0.0]# bin/kafka-topics.sh --zookeeper 192.168.231.250:2181 --list

testtopic

查询192.168.231.251上的topic

[root@kafka-1 kafka_2.11-2.0.0]# bin/kafka-topics.sh --zookeeper 192.168.231.251:2181 --list

testtopic

模拟消息生产和消费

发送消息到192.168.231.249

[root@kafka-1 kafka_2.11-2.0.0]# bin/kafka-console-producer.sh --broker-list 192.168.231.249:9092 --topic testtopic  
>nihao!

从192.168.231.250接受消息

[root@kafka-2 kafka_2.11-2.0.0]#  bin/kafka-console-consumer.sh --bootstrap-server  192.168.231.249:9092 --topic testtopic --from-beginning 

nihao!

注意：验证时如果报错，先检查是否后台运行的zk集群与kafka终止了。

查看全文

http://www.kler.cn/a/146905.html

【分割评价指标-nnUNet V2训练】- AutoDL

Java-异步方法@Async+自定义分布式锁注解Redission例子

.NET 简介

无人机在森林中的应用！

基于Python的仓库管理系统设计与实现

FPGA开发-逻辑分析仪的应用-数字频率计的设计

ZKP11.4 Use CI to instantiate Fiat-Shamir

探索编程在现代社会的无限价值

12 网关实战：Spring Cloud Gateway基础理论

Python break用法详解

[socket 弹 shell] msg_box3

虹科干货 | 适用于基于FPGA的网络设备的IEEE 1588透明时钟架构

线上异步任务突然不能回写100%

FLASK博客系列9——你想成为我的新用户吗？

Node.js入门指南（五）

vue页面表单提交时如何做校验

你了解Redis 的二进制安全吗

JS 单线程还是多线程，如何显示异步操作

Linux篇：文件系统

autojs-练手-批量下载JK图片

Windows bat批处理zip打包排除bat文件自身不打包

10、静态数码管显示

vatee万腾的科技征途：Vatee独特探索的数字化力量

excel单元格内换行按什么快捷键

【Spring整合Junit】Spring整合Junit介绍

边缘计算网关：智能制造的“智慧大脑”

kafka简介：

部署kafka

安装配置jdk8

安装配置ZK

安装

配置

配置项含义

创建数据日志目录

配置kafka

配置

配置项含义

创建log目录

启动，验证ZK集群

启动

验证，查看ZK配置

查看ZK状态

查看端口

启动验证kafka

启动

验证

在192.168.231.249创建topic

查询192.168.231.249上的topic

查询192.168.231.250上的topic

查询192.168.231.251上的topic

模拟消息生产和消费

发送消息到192.168.231.249

从192.168.231.250接受消息

注意：验证时如果报错，先检查是否后台运行的zk集群与kafka终止了。

相关文章：