Kafka环境搭建及使用教程

admin

kafka里自带了一个zookeeper，可以不需要再去下载zookeeper。在启动kafka前，需要先启动zookeeper。

（1）首先去修改config目录下的zookeeper.properties
（2）修改config目录下的server.properties，修改log.dirs和zookeeper.connect。log.dirs是日志存放文件夹，zookeeper.connect是zookeeper连接地址（端口和clientPort保持一致）。

* V/ O8 D1 l4 t9 W1 K1 L5 f: c& T至此，kafka单机版环境已搭建完毕。（如果对里面的参数没修改的话，可不修改，直接用默认的）

2 P) I* O; V5 S! k9 r$ s0 B; }

- x; A0 q7 n1 E2 ikafka使用教程
（1）启动kafka自带的zookeeper(加了-daemon，可以让这个进程挂在后台进行，否则会发现是个阻塞进程，只能再开一个终端进行其他操作)
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
" e8 Q! s1 |; z% ?0 J
（2）启动kafka
bin/kafka-server-start.sh -daemon config/server.properties

（3）创建一个topic
" S9 G; @: b6 ^# H7 G# _bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test


（4）查看topic列表
bin/kafka-topics.sh -list -zookeeper localhost:2181
6 Y" {' ~$ e1 T* Z3 Q2 O: X7 P
（5）创建生产者进程  （9092是默认的broker-list端口）
  r" Q4 e+ I; d- ^% J6 L  Z+ k6 bbin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
2 c; T6 h5 i5 Z! \2 V
2 a* k8 Y9 w4 l+ i, B! N, C$ f7 T（6）创建消费者进程
0 [' E9 ]. ]/ R7 M3 }! p, z% Fbin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

（7）外网访问 vi config/server.properties
advertised.listeners=PLAINTEXT://192.168.8.65:9092

admin

cat config/zookeeper.properties
  q" Q0 O9 r9 n8 z/ D6 f2 k# Licensed to the Apache Software Foundation (ASF) under one or more
* b9 l% d$ H! j( I' ?: [9 j' G  `# contributor license agreements.  See the NOTICE file distributed with
! l( ~, d/ Q0 L( @. {+ I# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
& y* e; y% G/ X& F7 X( _# the License.  You may obtain a copy of the License at
$ Z% A+ l$ u( q" N2 x#
#    http://www.apache.org/licenses/LICENSE-2.0
#
. u! }6 r! ~6 L# Unless required by applicable law or agreed to in writing, software
1 q6 v2 F; A5 M+ K, ^# distributed under the License is distributed on an "AS IS" BASIS,
5 d  g0 U* @6 \) \6 w0 @2 K$ u* A3 E# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
5 T3 |7 S' S3 s) R6 o' y/ w# See the License for the specific language governing permissions and
# limitations under the License.
# the directory where the snapshot is stored.
$ b! U6 x: n% U% J9 m" t% o3 V1 JdataDir=/mnt/kafka_2.13-3.3.1/data/zookeeper
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
5 [" H+ }. z: C2 }  R9 bmaxClientCnxns=0
# Disable the adminserver by default to avoid port conflicts.
# Set the port to something non-conflicting if choosing to enable this
4 v4 x6 X  c2 P. i" A: vadmin.enableServer=false
# admin.serverPort=8080


9 m: }7 M; G+ Q: m5 Z! ^4 Dnohup bin/zookeeper-server-start.sh config/zookeeper.properties &

admin

[root@ceph2 kafka_2.13-3.3.1]# egrep -v "^#|^$" config/server.properties
broker.id=0
4 C$ _! F+ f2 r- ~, R  J3 ~listeners=PLAINTEXT://192.168.0.233:9092
7 m, n/ L! y7 u$ p! u5 Tnum.network.threads=3
) ~" u; F7 x; j6 F8 gnum.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
& l1 I" @8 i, @4 J7 |socket.request.max.bytes=104857600
& b- R# Y8 D& c4 blog.dirs=/var/log/kafka-logs
num.partitions=1
* `# _( ~9 t0 b1 ~num.recovery.threads.per.data.dir=1
2 R: r) e% c" j0 [& L' ~% M8 Yoffsets.topic.replication.factor=1
3 Q; v+ c/ Z. f) Y1 q9 v" Mtransaction.state.log.replication.factor=1
# I' G( h- N  @( V1 z6 ttransaction.state.log.min.isr=1
log.retention.hours=168
" c% j5 E4 K7 }% w  R/ Nlog.retention.check.interval.ms=300000
/ \6 u0 O% |, l* ^# R5 C: C# Szookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=18000
/ V+ t* ?5 [2 Ggroup.initial.rebalance.delay.ms=0


[root@ceph2 kafka_2.13-3.3.1]# nohup bin/kafka-server-start.sh config/server.properties &

3 E+ M, u6 l) n' {2 |
9 j- L( v- H' r, ]- Z( b8 h' ~2 x
netstat  -ntlp |grep java
0 `. W, w- I1 f' E, stcp6       0      0 192.168.0.233:9092      :::*                    LISTEN      21222/java         
tcp6       0      0 :::2181                 :::*                    LISTEN      20780/java         
tcp6       0      0 :::41467                :::*                    LISTEN      21222/java         
) h: n$ h0 J7 vtcp6       0      0 :::40571                :::*                    LISTEN      20780/java