Kafka环境搭建及使用教程

admin

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

$ @, y  \% t' W4 J/ \5 T& P% \（1）首先去修改config目录下的zookeeper.properties
  b/ d4 W) b7 s* }8 n（2）修改config目录下的server.properties，修改log.dirs和zookeeper.connect。log.dirs是日志存放文件夹，zookeeper.connect是zookeeper连接地址（端口和clientPort保持一致）。
. |6 S) r9 s. z
- A) F5 x. D" u4 C至此，kafka单机版环境已搭建完毕。（如果对里面的参数没修改的话，可不修改，直接用默认的）


9 _$ E% ]$ \8 M9 C% n0 n+ k" m2 }9 _
kafka使用教程
/ K! z+ q3 c! y: q, P. D, M（1）启动kafka自带的zookeeper(加了-daemon，可以让这个进程挂在后台进行，否则会发现是个阻塞进程，只能再开一个终端进行其他操作)
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

（2）启动kafka
6 P3 ~4 S5 E& p, vbin/kafka-server-start.sh -daemon config/server.properties

（3）创建一个topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
, }% q8 m9 y) k- n
' G) [5 U! u( L
+ P; x  A4 T+ e& P- |+ M（4）查看topic列表
bin/kafka-topics.sh -list -zookeeper localhost:2181

1 q) @" M( p( g2 h7 s; p+ |（5）创建生产者进程  （9092是默认的broker-list端口）
$ h2 J8 Q) }& bbin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

（6）创建消费者进程
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
  _7 k4 ^. }3 b- w* M' E' j
6 R- x: H" C3 G+ m, t+ H（7）外网访问 vi config/server.properties
advertised.listeners=PLAINTEXT://192.168.8.65:9092
0 E+ l$ r/ K4 V

admin

cat config/zookeeper.properties
# Licensed to the Apache Software Foundation (ASF) under one or more
, X) J* U4 z! i) [$ _* Q# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
& Y2 A2 F1 t; Y9 Y# (the "License"); you may not use this file except in compliance with
* c8 o7 _0 W! U5 g! X' E, X2 ~# the License.  You may obtain a copy of the License at
#
0 G( J! T4 [" G( Z5 ]) n1 H! F#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
  g+ Y' N) o! L! j: Z2 d3 v5 e* [( U# distributed under the License is distributed on an "AS IS" BASIS,
( f1 e& k. e) v5 `# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
" c! G" h* N0 P# ?- Y" Z# the directory where the snapshot is stored.
* G& S6 m  U+ z3 LdataDir=/mnt/kafka_2.13-3.3.1/data/zookeeper
6 }% S2 q0 z$ f0 r9 T# the port at which the clients will connect
clientPort=2181
9 e) J1 Z; K: j0 ~$ ]. z* V# disable the per-ip limit on the number of connections since this is a non-production config
3 o7 y* Z. K1 _4 a( r0 G0 ?maxClientCnxns=0
( s% G6 j8 l$ I4 I  D6 m% }  Z# Disable the adminserver by default to avoid port conflicts.
; e- l' X# w! E# Set the port to something non-conflicting if choosing to enable this
# C# g( G& w# }5 R- O% E+ {/ @admin.enableServer=false
9 K' }4 D6 d7 z$ F# admin.serverPort=8080
3 s9 j0 `, M6 U4 F, `; [7 a
1 U  M5 i' `5 Z; O1 g$ w
nohup bin/zookeeper-server-start.sh config/zookeeper.properties &

admin

[root@ceph2 kafka_2.13-3.3.1]# egrep -v "^#|^$" config/server.properties
6 R1 r" V  F" [9 ~. ~1 [broker.id=0
% j$ \% v9 W8 K* nlisteners=PLAINTEXT://192.168.0.233:9092
num.network.threads=3
7 C$ T3 Y5 N5 |# X; ], Qnum.io.threads=8
) a5 J* I, W0 ]0 M" w% N, X) {# N& |socket.send.buffer.bytes=102400
, N. v+ A2 d$ y# R: rsocket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
& J; h0 k9 }, j* {3 R$ z' I# Tlog.dirs=/var/log/kafka-logs
num.partitions=1
2 U* T7 ]' E& P! j7 Z! a  V+ Hnum.recovery.threads.per.data.dir=1
" o& W7 t) x6 koffsets.topic.replication.factor=1
0 F, C7 Q5 _2 N! xtransaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
7 W1 {  @; A) X' e2 L) Y& a3 Czookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

# |. L$ a+ X6 \2 A" ]0 B) h
[root@ceph2 kafka_2.13-3.3.1]# nohup bin/kafka-server-start.sh config/server.properties &
5 B7 c) W5 x0 t0 a4 N
! q1 h, Z4 P9 ]# l
% _8 S* p% @  @; {  c% y
7 O) F+ ?: f: y' j" i/ mnetstat  -ntlp |grep java
: }& J! L& J: wtcp6       0      0 192.168.0.233:9092      :::*                    LISTEN      21222/java         
' h! T3 f% l8 Itcp6       0      0 :::2181                 :::*                    LISTEN      20780/java         
" ^2 L, l3 f' B1 z! gtcp6       0      0 :::41467                :::*                    LISTEN      21222/java         
tcp6       0      0 :::40571                :::*                    LISTEN      20780/java         

$ p1 @) n; F- C( d