将设为首页浏览此站
开启辅助访问 天气与日历 收藏本站联系我们切换到窄版

易陆发现论坛

 找回密码
 开始注册
查看: 37|回复: 2
收起左侧

Kafka单机模式和集群模式环境搭建

[复制链接]
发表于 2022-12-30 09:02:58 | 显示全部楼层 |阅读模式

马上注册,结交更多好友,享用更多功能,让你轻松玩转社区。

您需要 登录 才可以下载或查看,没有帐号?开始注册

x
单节点服务& a5 @% O% }, W6 x1 ]
1、下载解压:% w, l% l2 U1 V' p) }1 o9 Y
http://mirrors.tuna.tsinghua.edu ... afka_2.11-2.2.1.tgz
* |, J+ x. }7 L' [: k# D/ H/ P; B& i% p6 G& p1 l
2、启动服务* z' [/ J+ t$ ~; s/ Y4 f2 S
需要先启动ZooKeeper服务1 F3 W2 ~4 X  M% Z9 _
) t% Z$ b6 M& C3 o6 @
$ zkServer.sh start. e: @3 g" R7 n9 Y5 l. ~

2 E0 q* W% @6 K( ?8 c# 如果没有,可以使用单节点的ZooKeeper
2 R, H' Y/ [9 a& F; U. F$ bin/zookeeper-server-start.sh config/zookeeper.properties
1 m6 G; q' c  H: G9 g( P: j" _3 D4 h3 X7 G3 y6 q
启动Kafka, b2 [0 \8 n# i2 D

  M1 ?+ V9 U! _# M3 T1 m2 H$ F6 V> bin/kafka-server-start.sh config/server.properties! `* y' a+ U' ]; O

& V! n0 _* W/ L9 X& R! @$ A# s9 Y$ W3、创建话题7 s) r# k" V# w2 e
" u7 K, f; l1 X( C0 |; X
$ bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
) F0 @* y7 l% Y* K1 l8 y! N- i
/ E% d# ]% s- m0 \# 查看话题
3 S/ S+ C% _3 V2 G- [6 x$ X$ bin/kafka-topics.sh --list --bootstrap-server localhost:9092
/ e# F  J" }7 p+ Q9 _' ]( |9 H
/ ]0 P/ s" w, }2 I* T. ~" p2 ?1 z# L% r; a
4、发送消息, U8 C1 w- e, X5 I6 F* ~; W
4 u5 V3 Z! e) e
$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
& M/ A6 r, W# ^4 ^; D# ~This is a message4 ^2 p/ K$ C4 i) Z3 v/ U  c" k
This is another message4 a5 d2 {  o1 l  z4 |

# N' m) ~  S' C# w! f1 c! V5、接收消息
  r3 i% V& ]1 Z2 T- ~8 W
5 l- X2 M9 N6 A# ^$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning& i" ?+ w0 F" S3 _( q+ c
This is a message! O1 C" t6 M$ {& ^/ t& Q
This is another message* h3 y7 Y1 a# v# o$ v4 A

1 Q& C3 ?# m& B  Y; g: G多节点服务+ m( F) ]/ {+ z4 i3 \
1、修改配置) l& H- o% ^* e& j4 _; n
broker.id 在集群中需要唯一8 r, T6 n  Q, `5 m0 p0 I
& @& b; U/ G8 W4 i, J& o% J
$ cp config/server.properties config/server-1.properties
' x* u# K6 i& W* I1 N$ cp config/server.properties config/server-2.properties$ p& [- N+ o% I

! M5 d4 W; R; q  X; s5 ]. v! ?$ cat config/server-1.properties:$ ]8 F( W( x+ F* p7 H
broker.id=1) K, ~" D2 k! S( y5 }4 H6 W) j
listeners=PLAINTEXT://:9093
% h" s# Z' z& [; G. ^9 w4 }log.dirs=/tmp/kafka-logs-1* A4 b* H, p7 j3 a4 s& i
! l) j: z' d. J2 |/ @7 i
$ config/server-2.properties:& ^3 L/ l$ l8 w& \
broker.id=2
  j$ k0 E! t2 Z) J2 ~listeners=PLAINTEXT://:9094
/ x3 \; ]1 c/ o1 u& M, ~log.dirs=/tmp/kafka-logs-22 u+ U; P' E8 p3 u
4 }3 Y/ ~% P- y, S4 N3 C
2、创建话题- F3 F! r" h5 r6 Q  Q" C; f$ p

5 ~) r9 b" u- B4 z( }# R. [9 j6 _> bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 1 --topic my-replicated-topic
) P' I9 R9 L4 b4 y. i2 _  \4 l- y2 G
# 查看话题描述
0 J( k$ j- k* ~2 W9 [, l# k* f> bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic test# v4 Q& j% L3 G- _
4 J  R( `2 N& L: n, w! {1 b
> bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic my-replicated-topic8 g6 a8 Q# \' o; d# f; Q  z
, Q* H& E6 z2 y5 S- b, Q
3、查看进程,一共有6个(我的天)6 _" S$ s3 i: S* I1 c
! v5 Q0 P; L( R$ k& W/ {
登录后复制 - a* A$ g; [0 g5 n7 ^( X
$ jps9 C9 ~0 o. H; c! k5 ~
72513 Kafka             # 3个Kafka组成集群$ i  z" ~1 P$ ^' s5 d" W- q) c' Z7 P7 p
72816 Kafka
" ?" ]) u" b% s2 B: F2 Y73081 Kafka 8 J3 a: [+ a# T  ^. R( @
73872 Jps
, |0 @$ `0 E6 f2 q$ B* ^- |) q73347 ConsoleProducer   # 生产者3 t% M8 F- o* s9 h# ?! s- }; n
73609 ConsoleConsumer   # 消费者' `: w4 I& q1 `4 i6 |( c$ e) d
35198
  N4 i) h- [# }- \6 _# w& q68590 QuorumPeerMain    # Zookeeper
" O) _$ N5 y8 G6 C
7 W3 J% y& F  t% @8 v1 E' f: L" V1 j: u" t
 楼主| 发表于 2022-12-30 09:04:33 | 显示全部楼层
2、启动服务5 }; ^/ v3 A4 d5 j
需要先启动ZooKeeper服务; F6 f  _+ W0 _9 {1 p
$ zkServer.sh start; q; H* b, F# N3 k% M
# 如果没有,可以使用单节点的ZooKeeper% e& _0 n% l5 F0 p4 X
$ bin/zookeeper-server-start.sh config/zookeeper.properties
7 k4 J, b6 Z" E  ~# f启动Kafka
+ X. j. |) f+ f> bin/kafka-server-start.sh config/server.properties2 L  v* s- k- p3 R
3、创建话题
8 t# F' x& Z; d" }4 R! ^6 A/ X$ bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
7 i+ k  d( ]- z  ?7 [9 b6 K# 查看话题7 p  l6 r/ }( m+ v6 C' j/ |
$ bin/kafka-topics.sh --list --bootstrap-server localhost:9092
* u% ~: k9 p- `3 ?/ i4、发送消息
# }( A9 g) y: t! }. L# p$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
& p, _% J4 M1 C1 `2 PThis is a message) c9 W: L* W- b7 w. _8 E' c2 E' f
This is another message5 A/ V+ L3 E1 P& p2 E5 H2 i0 \
5、接收消息$ S8 ?+ \6 `) x
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
9 }0 t7 X$ `5 p& }" z; z8 cThis is a message
. y/ P; @% O6 N% V9 _7 hThis is another message
$ X- s2 y' t8 x; J7 B多节点服务  Z0 E0 ]' s# a/ ^- d0 w# w. D. f
1、修改配置
+ I- R6 W1 I. F! D6 }) X3 v1 ]' Ybroker.id 在集群中需要唯一1 x- t5 f+ k9 Z& F) u4 [
$ cp config/server.properties config/server-1.properties+ u3 O7 k; U' B
$ cp config/server.properties config/server-2.properties4 r& e; m/ p3 J" n  d2 r
$ cat config/server-1.properties:
- D, S' d! f" c( Vbroker.id=1; Z' C& G2 j! r8 @3 Z& a
listeners=PLAINTEXT://:90931 X2 S. _. O! }' I! _* i: R' _& X
log.dirs=/tmp/kafka-logs-1  t1 i0 N( F3 P5 i( U6 T% S% {

# B' ~* P1 t, d% t; O, L7 f$ config/server-2.properties:
& [. f" K9 P  _0 [6 a$ Q3 H6 Jbroker.id=26 c* L* \% D; i. C6 |" F2 F
listeners=PLAINTEXT://:9094
' e, m; S; |6 m& r7 T4 s! _. w  Blog.dirs=/tmp/kafka-logs-2
/ h  k/ Q5 k) H3 l4 t) m2、创建话题
3 m" i3 e" D' ^) |> bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 1 --topic my-replicated-topic
, \7 x! T8 a; l' c# |0 I/ N# 查看话题描述5 ^4 ~- g: u4 @/ z7 n
> bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic test/ E. m) q$ I3 x! [1 C) {
> bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic my-replicated-topic$ k: b3 ?+ N0 v/ I
3、查看进程,一共有6个(我的天)
8 ?) r" W. H) n2 ]' ?1 e% H4 d$ jps2 w* `  \2 a$ j
72513 Kafka             # 3个Kafka组成集群8 m8 t* M1 A7 r8 n% |1 a
72816 Kafka% V/ ^- O' q4 Z2 ?) L
73081 Kafka 1 M5 x# P. l. W/ @
73872 Jps* U3 L3 y8 Q. L4 k3 w
73347 ConsoleProducer   # 生产者
- P8 n; {" G! N% x5 i73609 ConsoleConsumer   # 消费者
7 l6 i5 P  a- m3 P351984 b- L6 A! t$ |6 Q% f( ~
68590 QuorumPeerMain    # Zookeeper
 楼主| 发表于 2022-12-30 09:07:15 | 显示全部楼层
在了解完 Kafka 的基本概念之后,我们通过搭建 Kafka 集群来进一步深刻认识一下 Kafka。$ \+ s  q8 x0 Q# @! V+ q
" Z2 P. I) V' [! P$ X1 _
确保安装环境, @8 B! ~7 @# [
安装 Java 环境
# ~" O$ k) A3 O在安装 Kafka 之前,先确保Linux 环境上是否有 Java 环境,使用 java -version 命令查看 Java 版本,推荐使用Jdk 1.8 ,如果没有安装 Java 环境的话,可以按照这篇文章进行安装(https://www.cnblogs.com/zs-notes/p/8535275.html& K2 m* ~# q) v8 {
* v% y1 c+ z2 s, U* a# \
安装 Zookeeper 环境
2 D/ x2 s, w6 i3 @Kafka 的底层使用 Zookeeper 储存元数据,确保一致性,所以安装 Kafka 前需要先安装 Zookeeper,Kafka 的发行版自带了 Zookeeper ,可以直接使用脚本来启动,不过安装一个 Zookeeper 也不费劲+ [! ?4 c! }2 S0 p
; C/ l3 R8 [4 [( s# k
Zookeeper 单机搭建
! A' f( ~5 `2 b8 A. w* L7 Y% `& r7 ZZookeeper 单机搭建比较简单,直接从 https://www.apache.org/dyn/closer.cgi/zookeeper/ 官网下载一个稳定版本的 Zookeeper ,这里我使用的是 3.4.10,下载完成后,在 Linux 系统中的 /usr/local 目录下创建 zookeeper 文件夹,使用xftp 工具(xftp 和 xshell 工具都可以在官网 https://www.netsarang.com/zh/xshell/ 申请免费的家庭版)把下载好的 zookeeper 压缩包放到 /usr/local/zookeeper 目录下。4 T1 q" I$ b% C7 |# i

# q- Y' q6 c7 U1 J& W6 P' k如果下载的是一个 tar.gz 包的话,直接使用 tar -zxvf zookeeper-3.4.10.tar.gz解压即可9 v8 ?, S& k# M" ?

! M6 z4 o# Y  z( R如果下载的是 zip 包的话,还要检查一下 Linux 中是否有 unzip 工具,如果没有的话,使用 yum install unzip 安装 zip 解压工具,完成后使用 unzip zookeeper-3.4.10.zip 解压即可。
; l4 x9 @) a% A, ]; x4 s9 U" U( b- b
解压完成后,cd 到 /usr/local/zookeeper/zookeeper-3.4.10 ,创建一个 data 文件夹,然后进入到 conf 文件夹下,使用 mv zoo_sample.cfg zoo.cfg 进行重命名操作
: e, U: g, b( K& }! G2 @) o5 r
0 n" D; z5 D3 a, G4 h1 S然后使用 vi 打开 zoo.cfg ,更改一下dataDir = /usr/local/zookeeper/zookeeper-3.4.10/data ,保存。
& `  D$ Q9 h* n) U+ f
  t1 ^2 a6 T4 r: r进入bin目录,启动服务输入命令 ./zkServer.sh start 输出下面内容表示搭建成功, X7 ?& U0 G2 g8 v$ T* |
( n: Z: `% y  f1 v
关闭服务输入命令,./zkServer.sh stop! J" }9 p# P6 C/ |5 s% M

' S2 B9 N; M% k( v( Q! \" ?* @" {$ H# F9 [
2 u1 k( C  W; j# B' S+ U
使用 ./zkServer.sh status 可以查看状态信息。
4 E$ g9 J/ W' M; x: `' \( h. j
) A' O) X$ w. v% f& ~Zookeeper 集群搭建8 }% s) Q% U; U5 j+ |' i
准备条件
- u' X) b6 F. A# C8 s2 }7 s准备条件:需要三个服务器,这里我使用了CentOS7 并安装了三个虚拟机,并为各自的虚拟机分配了1GB的内存,在每个 /usr/local/ 下面新建 zookeeper 文件夹,把 zookeeper 的压缩包挪过来,解压,完成后会有 zookeeper-3.4.10 文件夹,进入到文件夹,新建两个文件夹,分别是 data 和 log 文件夹
% I9 W) {3 a6 |- J8 v0 V# P" i  u- I! g
注:上一节单机搭建中已经创建了一个data 文件夹,就不需要重新创建了,直接新建一个 log 文件夹,对另外两个新增的服务需要新建这两个文件夹。
! _8 n7 H  Z  R
) S% n6 M  q2 S+ U设置集群
7 K7 E: ?! ^/ X新建完成后,需要编辑 conf/zoo.cfg 文件,三个文件的内容如下
: O  j6 c+ N' l) f) m6 q4 z" u  |  y( s
tickTime=2000
4 k; Q2 V# W+ Y9 B' }initLimit=100 J6 m1 V* z9 n/ s, u; f9 d% d
syncLimit=5
! ?9 u2 F7 R. ndataDir=/usr/local/zookeeper/zookeeper-3.4.10/data/ l& h% i9 c- W1 U! t
dataLogDir=/usr/local/zookeeper/zookeeper-3.4.10/log
# B3 c$ Y* l2 f2 bclientPort=12181
% Y2 N: E7 n0 f5 c+ D# s+ ^0 |server.1=192.168.1.7:12888:13888
$ s& m0 F0 m2 userver.2=192.168.1.8:12888:13888  C- L+ |- _$ M* T0 R9 `
server.3=192.168.1.9:12888:13888
0 N1 P( [9 E4 ]2 j) J/ ?server.1 中的这个 1 表示的是服务器的标识也可以是其他数字,表示这是第几号服务器,这个标识要和下面我们配置的 myid 的标识一致可以。
% u. G8 w) {+ M* ]0 F9 I
- k( c7 d+ {1 W( d7 N5 }( I) Y- k192.168.1.7:12888:13888 为集群中的 ip 地址,第一个端口表示的是 master 与 slave 之间的通信接口,默认是 2888,第二个端口是leader选举的端口,集群刚启动的时候选举或者leader挂掉之后进行新的选举的端口,默认是 3888. u( K8 V( B/ K& Q

7 ^  |+ a; E$ [& {现在对上面的配置文件进行解释* Z9 v8 h: d9 j) h2 L! E  L$ i

+ [% M2 ?+ t8 e7 ^, RtickTime: 这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔,也就是每个 tickTime 时间就会发送一个心跳。9 F! X' C' z8 }8 ^

8 K& ~$ s) W; I3 ]. z; m* U: V0 q, G: a5 winitLimit:这个配置项是用来配置 Zookeeper 接受客户端(这里所说的客户端不是用户连接 Zookeeper 服务器的客户端,而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器)初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过 5个心跳的时间(也就是 tickTime)长度后 Zookeeper 服务器还没有收到客户端的返回信息,那么表明这个客户端连接失败。总的时间长度就是 5*2000=10 秒
, M( _8 u2 j$ J" z8 c1 ]( k+ d! v1 R' p" W. L
syncLimit: 这个配置项标识 Leader 与Follower 之间发送消息,请求和应答时间长度,最长不能超过多少个 tickTime 的时间长度,总的时间长度就是5*2000=10秒
% R6 h8 z/ B% x5 G; G2 E9 Y$ z5 K$ _# I
dataDir: 快照日志的存储路径2 `* T4 I5 s# Z+ ?2 a4 G

1 o3 \4 Z8 }! l) x8 a1 t/ adataLogDir: 事务日志的存储路径,如果不配置这个那么事务日志会默认存储到dataDir指定的目录,这样会严重影响zk的性能,当zk吞吐量较大的时候,产生的事务日志、快照日志太多
0 W6 Q5 m, l5 M0 V( N" K; Y. M/ r; v; ^" d% f2 p7 F
clientPort: 这个端口就是客户端连接 Zookeeper 服务器的端口,Zookeeper 会监听这个端口,接受客户端的访问请求。7 I' `8 l! W6 I* Y8 F
0 q( E" R* R( f. o7 s( `! d
创建 myid 文件3 d, h, F" r: g
在了解完其配置文件后,现在来创建每个集群节点的 myid ,我们上面说过,这个 myid 就是 server.1 的这个 1 ,类似的,需要为集群中的每个服务都指定标识,使用 echo 命令进行创建7 v- h! f+ ?! T& n, A" Z3 \

2 D) J; _4 O% Q# X4 k* K: f# server.1
. h# ?" i+ {1 X; l& kecho "1" > /usr/local/zookeeper/zookeeper-3.4.10/data/myid" ^5 J. h8 g  M
# server.2
, W) n. ]9 M+ u. ~9 s5 n4 h9 kecho "2" > /usr/local/zookeeper/zookeeper-3.4.10/data/myid9 @  g0 T+ Y" O1 ~/ o
# server.3
, b. A4 o) W5 A! t8 S' Decho "3" > /usr/local/zookeeper/zookeeper-3.4.10/data/myid2 p* f2 ]. X" L( S, M
启动服务并测试
+ \3 M! I% [0 B' V配置完成,为每个 zk 服务启动并测试,我在 windows 电脑的测试结果如下3 D& D( M  P4 N: b) Y

; m! J- Q$ M- `$ n启动服务(每台都需要执行): t: t7 I0 \; m  A: v" n
. Z: |0 O1 }) b
cd /usr/local/zookeeper/zookeeper-3.4.10/bin: f! Y" e. u) P2 U8 n/ l( x
./zkServer.sh start1 T3 k( ?8 Z. ~+ x# ]' k& u
检查服务状态+ L0 W  c) q( T" `5 m3 J

$ E  R  y2 o) t* j! e使用 ./zkServer.sh status 命令检查服务状态% T+ T- j' b" H, {0 I$ d8 `
9 `7 \: K3 [; e* \) V! _& v& r; N
192.168.1.7 — follower
, _: I1 b- ?( t4 V. N9 V! F/ u. H& j5 s" Q: g  ]1 A

8 v; H2 `4 `$ L# x( b  E# j8 t/ j- |7 r2 p$ y
192.168.1.8 — leader( t6 U; j$ N( _6 h
' I; V! }$ O: e( Y" }6 C

5 o3 w0 S* @, C" k: a: h5 @& c7 N8 O  y& W
192.168.1.9 — follower4 I) m% m" g7 e( ]! |# L
, c6 h9 V6 t/ D' s! t- K8 K

" }5 p' I% B4 M3 e; I1 D
8 }8 |: H+ J2 I0 W/ gzk集群一般只有一个leader,多个follower,主一般是相应客户端的读写请求,而从主同步数据,当主挂掉之后就会从follower里投票选举一个leader出来。
# I5 O/ F2 `( f5 Q
& q6 t! k+ f* s: \Kafka 集群搭建6 M) I; P7 O, ~0 V( d
准备条件
' D9 s! P, d! `% L3 S! D搭建好的 Zookeeper 集群
9 B7 p- N) f' w: aKafka 压缩包 (https://www.apache.org/dyn/close ... afka_2.12-2.3.0.tgz
+ l7 N. p4 n  |2 H在 /usr/local 下新建 kafka 文件夹,然后把下载完成的 tar.gz 包移到 /usr/local/kafka 目录下,使用 tar -zxvf 压缩包 进行解压,解压完成后,进入到 kafka_2.12-2.3.0 目录下,新建 log 文件夹,进入到 config 目录下" \2 J; ]  V$ ^& S1 l, V2 T
$ T. f  H* ^3 y
我们可以看到有很多 properties 配置文件,这里主要关注 server.properties 这个文件即可。
! ~5 ^! |8 G- R4 q% l2 H0 k2 `3 \- L' Q/ V# s( k' @( w! ~

) [: z8 H8 T% P2 K2 d. g" Z# f7 M2 }% Z1 c( f: o
kafka 启动方式有两种,一种是使用 kafka 自带的 zookeeper 配置文件来启动(可以按照官网来进行启动,并使用单个服务多个节点来模拟集群http://kafka.apache.org/quickstart#quickstart_multibroker),一种是通过使用独立的zk集群来启动,这里推荐使用第二种方式,使用 zk 集群来启动9 f7 c$ ~; I9 i. _8 r) f1 K

- g4 _: \$ R- n修改配置项
# b+ b0 L" l( x8 b需要为每个服务都修改一下配置项,也就是server.properties, 需要更新和添加的内容有9 B6 U- w4 R: M' ?/ D
  L& r# f! [* f' P% e
broker.id=0 //初始是0,每个 server 的broker.id 都应该设置为不一样的,就和 myid 一样 我的三个服务分别设置的是 1,2,3
8 {# Z& m! M" u! r; [7 M, \log.dirs=/usr/local/kafka/kafka_2.12-2.3.0/log$ |# g+ R) A/ t- _

# h+ X8 q( C9 \$ H! p0 D#在log.retention.hours=168 下面新增下面三项8 t& C/ Q/ |) l8 i$ j9 D: U6 h
message.max.byte=52428806 u# ^3 S9 [! `1 ?5 E2 h. s
default.replication.factor=2
% M1 |* k6 C( f3 |: O. _4 B  Jreplica.fetch.max.bytes=5242880
; @4 v  o) x$ x% U1 Z& k/ m
; m; V7 D; e9 W* T  F0 {: l; K8 {  ]#设置zookeeper的连接端口
) J/ i- x- W! E8 a) e: g0 dzookeeper.connect=192.168.1.7:2181,192.168.1.8:2181,192.168.1.9:2181
  Y1 C" {# c5 A3 ]: n7 k配置项的含义
' Z/ D- {+ ?9 s$ H6 ^8 W* V
9 f$ v6 U) a8 R, j$ P/ Abroker.id=0  #当前机器在集群中的唯一标识,和zookeeper的myid性质一样
' @) p& h' c7 z3 ^$ F" A# {9 Mport=9092 #当前kafka对外提供服务的端口默认是9092( ^0 v* J. c3 q+ B
host.name=192.168.1.7 #这个参数默认是关闭的,在0.8.1有个bug,DNS解析问题,失败率的问题。7 o; h' ]6 u3 Y5 w- Q
num.network.threads=3 #这个是borker进行网络处理的线程数
( }/ D5 x" f- s8 x; Y' |8 U& fnum.io.threads=8 #这个是borker进行I/O处理的线程数+ N2 W% `* [. v/ k2 I
log.dirs=/usr/local/kafka/kafka_2.12-2.3.0/log #消息存放的目录,这个目录可以配置为“,”逗号分割的表达式,上面的num.io.threads要大于这个目录的个数这个目录,如果配置多个目录,新创建的topic他把消息持久化的地方是,当前以逗号分割的目录中,那个分区数最少就放那一个
) x0 _9 g# Q1 f; _0 J. Csocket.send.buffer.bytes=102400 #发送缓冲区buffer大小,数据不是一下子就发送的,先回存储到缓冲区了到达一定的大小后在发送,能提高性能* }  N( U4 n* I0 R# [3 u2 ]# W
socket.receive.buffer.bytes=102400 #kafka接收缓冲区大小,当数据到达一定大小后在序列化到磁盘
# x  R) L0 q" `$ csocket.request.max.bytes=104857600 #这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数,这个值不能超过java的堆栈大小) J* ^- Z/ }# `! ]2 c( O
num.partitions=1 #默认的分区数,一个topic默认1个分区数, L! s& q! s  J3 G+ E
log.retention.hours=168 #默认消息的最大持久化时间,168小时,7天
: ?' [$ M! Q; c) _! }message.max.byte=5242880  #消息保存的最大值5M% A% `  q. {! O, E% o4 ^
default.replication.factor=2  #kafka保存消息的副本数,如果一个副本失效了,另一个还可以继续提供服务' Z. Z8 W1 H2 z0 |- o
replica.fetch.max.bytes=5242880  #取消息的最大直接数
. g: t* J/ L! W: y9 c- p+ Hlog.segment.bytes=1073741824 #这个参数是:因为kafka的消息是以追加的形式落地到文件,当超过这个值的时候,kafka会新起一个文件
& G% c/ A6 }) T2 n- J6 p) Nlog.retention.check.interval.ms=300000 #每隔300000毫秒去检查上面配置的log失效时间(log.retention.hours=168 ),到目录查看是否有过期的消息如果有,删除
& z# T" C: b% S1 {/ Y. v" Jlog.cleaner.enable=false #是否启用log压缩,一般不用启用,启用的话可以提高性能
7 Q# O: E4 h0 _( Gzookeeper.connect=192.168.1.7:2181,192.168.1.8:2181,192.168.1.9:2181 #设置zookeeper的连接端口
& v/ r3 C2 m1 e. \9 t0 M% B启动 Kafka 集群并测试
+ N* s% z* c+ r- C! o启动服务,进入到 /usr/local/kafka/kafka_2.12-2.3.0/bin 目录下* o  ^9 q+ ^! P# r! \5 Z( e2 t4 z
# 启动后台进程0 u  j% a; v" N! O7 {; Y
./kafka-server-start.sh -daemon ../config/server.properties: C( Y# S* ^6 r7 u( o
检查服务是否启动
5 `0 E. B5 ]0 S/ g- r# 执行命令 jps$ ^; o  X( f7 N# N
6201 QuorumPeerMain
, A" e! e5 u  Q3 f6 i9 U/ J& p2 W+ I7035 Jps7 c& V8 }) U" w( ^& n* w2 |
6972 Kafka
3 ^% V; Z/ V3 U5 B  S- o8 akafka 已经启动
8 D. M0 k, u) @  [" W6 D$ r) B创建 Topic 来验证是否创建成功
; B- `1 ?2 _5 [4 I# cd .. 往回退一层 到 /usr/local/kafka/kafka_2.12-2.3.0 目录下( {+ k/ P$ l# L' M
bin/kafka-topics.sh --create --zookeeper 192.168.1.7:2181 --replication-factor 2 --partitions 1 --topic cxuan
" H6 p* ~" C! j8 `对上面的解释
" L* t3 a/ t9 H1 I& S+ ?$ ]
. _7 }  E! x" g–replication-factor 2 复制两份. O4 K4 p% ^7 N1 Y

4 n3 \5 ?, u, @4 m/ ^* [3 H–partitions 1 创建1个分区: e6 P8 x- u+ L" [- X9 L  Q
' m/ T  R: A. `% o
–topic 创建主题
" H2 }% E$ `4 y, H* m% x. Y
/ c; S* m; \% a2 |3 L/ m1 r; ?: \) K查看我们的主题是否出创建成功
5 g; K& W* L4 y6 C
% p) F# B5 t% X9 z$ h& Abin/kafka-topics.sh --list --zookeeper 192.168.1.7:2181' [5 U+ B9 n! W) w9 X' \$ z: @/ h

0 j  P4 H% g1 r, X; V7 L6 A2 b1 e" ?3 s9 X& L" V+ V# @0 ?) y( [
启动一个服务就能把集群启动起来- r, Y% V8 {; e  m' ^

) ?7 c# ~- L6 e  c在一台机器上创建一个发布者
) L7 P* `2 v% P* Q3 R  S2 g+ a( V. n6 S% |( r+ T. a
# 创建一个broker,发布者- Z! O. o! a! q0 `# `- C
./kafka-console-producer.sh --broker-list 192.168.1.7:9092 --topic cxuantopic5 K# ~( n) a4 Q5 |# h) i
在一台服务器上创建一个订阅者
& G' P( J  f8 C7 S, G; P$ I0 i$ Q
# 创建一个consumer, 消费者
; a: B. `  m9 l  x" }/ ^. g, Vbin/kafka-console-consumer.sh --bootstrap-server 192.168.1.7:9092 --topic cxuantopic --from-beginning
* r% S7 J! }, _! _+ ~注意:这里使用 –zookeeper 的话可能出现 zookeeper is not a recognized option 的错误,这是因为 kafka 版本太高,需要使用 --bootstrap-server 指令
# l5 A. u8 A+ X2 t+ R6 N; c4 o9 b1 [) y  V! g( O  K* T. y
测试结果
- @( z( Q2 h% k- I& h4 \6 l0 X* l; c) N: q
发布. y& ^* i! v; A; r: B, S7 k0 o- Z7 f

4 ?8 s3 l5 L& X4 U$ y, w
1 `: z0 t( J( q2 A# X- [, v" d! J( }% z2 {% n, u" q
消费
! S7 ]& k6 w' M' O. ^) r* x5 N7 k7 A5 m3 Y, d2 P  S6 i

; T% L- f4 W' @7 B( c9 v5 d+ I+ T" e: n) g
其他命令0 Y7 i% X* Z4 ^0 x5 k- Z
显示 topic
8 I( G- a$ ]3 U! M! n1 L" v
' m* Q" p- [5 a- [  vbin/kafka-topics.sh --list --zookeeper 192.168.1.7:2181
$ p& L2 U* M4 P- z5 R3 q9 i% f0 g* o1 O; J( d7 o
# 显示  g; h- v' e! c/ Z' Y4 l- b
cxuantopic
# a, [$ S! l3 ]查看 topic 状态
3 U5 d$ s0 H$ D8 G8 [( `7 G
* Z" P& g4 l& }2 U2 Sbin/kafka-topics.sh --describe --zookeeper 192.168.1.7:2181 --topic cxuantopic( Q; ^) p$ {$ p' N

* }1 g. A0 p, i7 n- C; K, E( X# 下面是显示的详细信息
( O4 C8 {4 a4 S2 PTopic:cxuantopic PartitionCount:1 ReplicationFactor:2 Configs:. E5 v( [; m2 [$ A0 l( V/ X
Topic: cxuantopic Partition: 0 Leader: 1 Replicas: 1,2 Isr: 1,2( i( s/ e3 M2 s; T1 j  B
3 ?# f1 d- M2 H3 b) [! I
# 分区为为1  复制因子为2   主题 cxuantopic 的分区为0
8 U* @# q. s" O/ Z# Replicas: 0,1   复制的为1,2
6 O- h, k" W& R  F) i/ J+ OLeader 负责给定分区的所有读取和写入的节点,每个节点都会通过随机选择成为 leader。
- ~" T* x. \% s. @" }) F
. `0 o! c6 P1 M8 `7 z3 N2 F4 YReplicas 是为该分区复制日志的节点列表,无论它们是 Leader 还是当前处于活动状态。4 g$ Q/ g! d! }5 A( }

) j3 l0 O9 a; G2 c+ ~Isr 是同步副本的集合。它是副本列表的子集,当前仍处于活动状态并追随Leader。
& D/ h4 Q8 N1 }; F0 c1 o) [+ X% g- T1 e& A
至此,kafka 集群搭建完毕。
: G1 I& E5 k( H* a2 @. J4 x7 A, _* r( u9 k) a$ s# S, A
验证多节点接收数据4 J- L0 I4 R+ R* O! _7 l9 L- A
刚刚我们都使用的是 相同的ip 服务,下面使用其他集群中的节点,验证是否能够接受到服务4 D& l* k/ C$ h0 e$ v
7 f: [9 @' }8 X! Q$ n$ z$ ]: h
在另外两个节点上使用* D+ g3 M1 i% I' E' I. T0 m

* V2 o$ X* K8 E& W# [0 M3 j9 `7 y  Dbin/kafka-console-consumer.sh --bootstrap-server 192.168.1.7:9092 --topic cxuantopic --from-beginning
5 f0 @/ Y1 e. l; P) N( |# _然后再使用 broker 进行消息发送,经测试三个节点都可以接受到消息。: i. K  h/ `8 p/ t1 z

' |+ X+ N  s, \- h2 C配置详解' c) F" b  o* m& V( @$ w
在搭建 Kafka 的时候我们简单介绍了一下 server.properties 中配置的含义,现在我们来详细介绍一下参数的配置和概念
4 G3 @0 y2 J1 s& t! v2 Z
8 x: s1 e- s5 \* h常规配置
3 O, A8 P+ M: A* n0 S4 S# k" O$ M8 f% F这些参数是 kafka 中最基本的配置& \; W! g# s/ S1 W( `  `# B

( b4 J6 |7 V$ f& K1 Qbroker.id& I" t7 ?- F/ k. U& [
每个 broker 都需要有一个标识符,使用 broker.id 来表示。它的默认值是 0,它可以被设置成其他任意整数,在集群中需要保证每个节点的 broker.id 都是唯一的。3 t$ G+ O1 k$ z8 N: i! q% s3 \& l
. M; n# @& R% X, c. }+ d
port
7 q( h8 P6 B& D& ~2 C' z: W如果使用配置样本来启动 kafka ,它会监听 9092 端口,修改 port 配置参数可以把它设置成其他任意可用的端口。  N9 p3 \, K2 A# u5 O% l

- F4 ^8 L! _( C& I4 I, tzookeeper.connect
, o, b7 |2 c* ?( F用于保存 broker 元数据的地址是通过 zookeeper.connect 来指定。 localhost:2181 表示运行在本地 2181 端口。该配置参数是用逗号分隔的一组 hostname:port/path 列表,每一部分含义如下:
$ w( @* ^$ I/ N2 X. v
' u! I6 N' @; K) t* T- |- chostname 是 zookeeper 服务器的服务名或 IP 地址
; ], s/ Z- ]' _1 M5 Q1 r
( z) `- c/ F5 P" [port 是 zookeeper 连接的端口
; H2 P. O9 H$ q- z' ]3 z1 C8 X8 F: e  }# a9 I% T
/path 是可选的 zookeeper 路径,作为 Kafka 集群的 chroot 环境。如果不指定,默认使用跟路径8 f8 u) f: s, v7 P) g0 E& X
0 T, T2 u2 W3 D, C5 I  {
log.dirs
$ ?- [3 S! z  j) }- sKafka 把消息都保存在磁盘上,存放这些日志片段的目录都是通过 log.dirs 来指定的。它是一组用逗号分隔的本地文件系统路径。如果指定了多个路径,那么 broker 会根据 “最少使用” 原则,把同一分区的日志片段保存到同一路径下。要注意,broker 会向拥有最少数目分区的路径新增分区,而不是向拥有最小磁盘空间的路径新增分区。
: E2 S- b6 Z+ E: e1 C
. g! h' X, o4 S9 A, T& d: {5 onum.recovery.threads.per.data.dir
1 ]6 k" ^% ?* p* L; W( |对于如下 3 种情况,Kafka 会使用可配置的线程池来处理日志片段
6 R  v) S1 l% F
0 F( F1 s# O/ _8 x5 J服务器正常启动,用于打开每个分区的日志片段;
! R. L: l+ \' y% [7 R  i
+ C/ {8 e/ T% c. I) ?- f% c9 V( t) e服务器崩溃后启动,用于检查和截断每个分区的日志片段;
) a! O. ]5 G1 m/ s5 R2 a3 ~
$ K9 k! g/ d2 ~/ w7 H服务器正常关闭,用于关闭日志片段
$ p; r- A# R% P! ]! q$ |
( e0 D* `2 z2 k# G默认情况下,每个日志目录只使用一个线程。因为这些线程只是在服务器启动和关闭时会用到,所以完全可以设置大量的线程来达到井行操作的目的。特别是对于包含大量分区的服务器来说,一旦发生崩愤,在进行恢复时使用井行操作可能会省下数小时的时间。设置此参数时需要注意,所配置的数字对应的是 log.dirs 指定的单个日志目录。也就是说,如果 num.recovery.threads.per.data.dir 被设为 8,并且 log.dir 指定了 3 个路径,那么总共需要 24 个线程。
" x. x9 T6 \+ I- {
, f& x3 Z+ L9 y) Rauto.create.topics.enable) J% a: s5 c$ g6 g5 }
默认情况下,Kafka 会在如下 3 种情况下创建主题
, N( z. l( ^/ Y% m% U4 Z/ C
; G% j6 f6 `! O& Z9 K当一个生产者开始往主题写入消息时* d2 @% I$ Q" l/ x7 F

2 b/ ?6 n6 T3 |8 {; p% c$ f  `( P/ O2 A当一个消费者开始从主题读取消息时* M& |0 r: ]8 k8 N! i& j
) g. ?) A' h1 u$ ?4 w, z9 a1 m. d
当任意一个客户向主题发送元数据请求时
1 @5 R# X3 n" ~3 s9 O$ t9 q$ ?+ d. Z9 P2 ]  o
delete.topic.enable9 d7 `) X0 Z& w, w! }# X
如果你想要删除一个主题,你可以使用主题管理工具。默认情况下,是不允许删除主题的,delete.topic.enable 的默认值是 false 因此你不能随意删除主题。这是对生产环境的合理性保护,但是在开发环境和测试环境,是可以允许你删除主题的,所以,如果你想要删除主题,需要把 delete.topic.enable 设为 true。
6 V4 |% B0 h  ~0 _( s1 ]! p1 d! g7 I0 ?1 o2 O) F
主题默认配置2 _" m5 r' L* h% }3 A4 y
Kafka 为新创建的主题提供了很多默认配置参数,下面就来一起认识一下这些参数
9 r3 Y/ U. w% h# l* k
7 e: `, y. }" B: `+ K- d# {2 Y: Tnum.partitions" q2 Y) L6 t0 d7 A- e
num.partitions 参数指定了新创建的主题需要包含多少个分区。如果启用了主题自动创建功能(该功能是默认启用的),主题分区的个数就是该参数指定的值。该参数的默认值是 1。要注意,我们可以增加主题分区的个数,但不能减少分区的个数。
- t: |; ~- m4 p
( V/ X  ~# \" D7 Y0 {: qdefault.replication.factor, P# N; e3 s& L; H
这个参数比较简单,它表示 kafka保存消息的副本数,如果一个副本失效了,另一个还可以继续提供服务default.replication.factor 的默认值为1,这个参数在你启用了主题自动创建功能后有效。
, ~  W4 R+ {4 U1 b( A+ K: ]  f
3 J+ c* [# S  blog.retention.ms, t; {8 S$ U" t1 E, N# _+ X
Kafka 通常根据时间来决定数据可以保留多久。默认使用 log.retention.hours 参数来配置时间,默认是 168 个小时,也就是一周。除此之外,还有两个参数 log.retention.minutes 和 log.retentiion.ms 。这三个参数作用是一样的,都是决定消息多久以后被删除,推荐使用 log.retention.ms。
. V2 _* V! Q% T$ X- u
6 l: J3 N2 {$ Y7 r( o0 f! K. X3 i- Olog.retention.bytes. `, }! @6 v& j$ N( h
另一种保留消息的方式是判断消息是否过期。它的值通过参数 log.retention.bytes 来指定,作用在每一个分区上。也就是说,如果有一个包含 8 个分区的主题,并且 log.retention.bytes 被设置为 1GB,那么这个主题最多可以保留 8GB 数据。所以,当主题的分区个数增加时,整个主题可以保留的数据也随之增加。, [, k3 g3 B: r

# g! s2 P$ W" blog.segment.bytes
  t0 X2 ?( r$ \* t上述的日志都是作用在日志片段上,而不是作用在单个消息上。当消息到达 broker 时,它们被追加到分区的当前日志片段上,当日志片段大小到达 log.segment.bytes 指定上限(默认为 1GB)时,当前日志片段就会被关闭,一个新的日志片段被打开。如果一个日志片段被关闭,就开始等待过期。这个参数的值越小,就越会频繁的关闭和分配新文件,从而降低磁盘写入的整体效率。. E6 S1 b% m4 M- `8 b
' T/ y$ G5 c8 Z. s+ _# Q
log.segment.ms1 J4 M+ V* }; F9 `! [. J& L
上面提到日志片段经关闭后需等待过期,那么 log.segment.ms 这个参数就是指定日志多长时间被关闭的参数和,log.segment.ms 和 log.retention.bytes 也不存在互斥问题。日志片段会在大小或时间到达上限时被关闭,就看哪个条件先得到满足。+ s& S( b- x' B% @$ e

! q& P# l; Z* P8 c, C  l0 G( `message.max.bytes
6 M/ d9 p- I9 [& dbroker 通过设置 message.max.bytes 参数来限制单个消息的大小,默认是 1000 000, 也就是 1MB,如果生产者尝试发送的消息超过这个大小,不仅消息不会被接收,还会收到 broker 返回的错误消息。跟其他与字节相关的配置参数一样,该参数指的是压缩后的消息大小,也就是说,只要压缩后的消息小于 mesage.max.bytes,那么消息的实际大小可以大于这个值
& a* G2 N* b" D0 {5 f. M
3 N' k7 p- i7 j9 O: V; |这个值对性能有显著的影响。值越大,那么负责处理网络连接和请求的线程就需要花越多的时间来处理这些请求。它还会增加磁盘写入块的大小,从而影响 IO 吞吐量。
您需要登录后才可以回帖 登录 | 开始注册

本版积分规则

关闭

站长推荐上一条 /4 下一条

如有购买积分卡请联系497906712

QQ|返回首页|Archiver|手机版|小黑屋|易陆发现 点击这里给我发消息

GMT+8, 2023-2-1 00:56 , Processed in 0.050221 second(s), 22 queries .

Powered by LR.LINUX.cloud bbs168x X3.2 Licensed

© 2012-2022 Comsenz Inc.

快速回复 返回顶部 返回列表