Kafka环境搭建及使用教程

admin

kafka里自带了一个zookeeper，可以不需要再去下载zookeeper。在启动kafka前，需要先启动zookeeper。
# B1 `5 P" A  t5 E9 q' \
（1）首先去修改config目录下的zookeeper.properties
（2）修改config目录下的server.properties，修改log.dirs和zookeeper.connect。log.dirs是日志存放文件夹，zookeeper.connect是zookeeper连接地址（端口和clientPort保持一致）。

至此，kafka单机版环境已搭建完毕。（如果对里面的参数没修改的话，可不修改，直接用默认的）
" X3 ~* C; B( p( w

) m" D% \: z2 S
; R2 O3 A$ L: `  g( }: c  okafka使用教程
（1）启动kafka自带的zookeeper(加了-daemon，可以让这个进程挂在后台进行，否则会发现是个阻塞进程，只能再开一个终端进行其他操作)
( x2 H1 Q3 r& u( p2 W9 h/ Hbin/zookeeper-server-start.sh -daemon config/zookeeper.properties

（2）启动kafka
& ^; d' T2 ~: Q6 ybin/kafka-server-start.sh -daemon config/server.properties
" G+ {! S4 Z' j0 d0 j
2 e) _  X* u3 `" A（3）创建一个topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test


（4）查看topic列表
2 o# o5 Q, v9 m* s8 A- Obin/kafka-topics.sh -list -zookeeper localhost:2181
2 v2 J  R2 r+ h
（5）创建生产者进程  （9092是默认的broker-list端口）
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
* t1 f6 z9 b( [1 F
（6）创建消费者进程
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

- T5 t5 ~4 l0 J（7）外网访问 vi config/server.properties
/ Y* ~: @" G( v+ s4 X! eadvertised.listeners=PLAINTEXT://192.168.8.65:9092

admin

cat config/zookeeper.properties
5 J" k3 N6 ~* y; f# Licensed to the Apache Software Foundation (ASF) under one or more
* T. C( _  l: V# 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
# (the "License"); you may not use this file except in compliance with
' F2 `+ x+ B/ f) y& i# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
2 l9 Q) K# o7 C+ F7 t, g! A/ B#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
9 {0 ?' H; \" y1 r. p& h# See the License for the specific language governing permissions and
; d1 Q5 G- U, D% E/ o% l0 Y7 b  Y# limitations under the License.
# the directory where the snapshot is stored.
dataDir=/mnt/kafka_2.13-3.3.1/data/zookeeper
8 i0 T- |! E, x, ?: e7 ~1 }: ?# the port at which the clients will connect
  P/ E" Y$ H9 T9 _7 B7 A( O  Y' d) \: AclientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
$ Y9 ]6 I) R7 |5 U: \maxClientCnxns=0
9 p) R$ {7 x+ q  _  q; u# Disable the adminserver by default to avoid port conflicts.
# Set the port to something non-conflicting if choosing to enable this
. b: z6 K" b9 x- E$ }admin.enableServer=false
# admin.serverPort=8080


nohup bin/zookeeper-server-start.sh config/zookeeper.properties &
& G) W( @8 z5 I& q! W: `: \9 q

admin

[root@ceph2 kafka_2.13-3.3.1]# egrep -v "^#|^$" config/server.properties
broker.id=0
listeners=PLAINTEXT://192.168.0.233:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
- f3 P( h4 Y- I; p: z0 v( r' _socket.receive.buffer.bytes=102400
8 x, @  K& Y8 _5 B8 @$ y( Dsocket.request.max.bytes=104857600
$ X9 d! l& l% _, l3 l, j* O( _, [log.dirs=/var/log/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
! n9 n5 y. N+ ]( x7 l' i9 ~$ T  s3 ]; Rtransaction.state.log.replication.factor=1
% X- c. t! S) B$ |2 X9 Htransaction.state.log.min.isr=1
. ?2 e' N8 V4 I* e* ilog.retention.hours=168
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
1 u& h- |. e, szookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0


4 c) e1 b' ^7 K; x1 P8 X& a[root@ceph2 kafka_2.13-3.3.1]# nohup bin/kafka-server-start.sh config/server.properties &


5 Q$ F9 q- v, h4 A! U: C% P+ d8 \
netstat  -ntlp |grep java
tcp6       0      0 192.168.0.233:9092      :::*                    LISTEN      21222/java         
* @% p/ G5 K9 Y- c! {' ~; Stcp6       0      0 :::2181                 :::*                    LISTEN      20780/java         
) ?- j1 n+ N) e8 Rtcp6       0      0 :::41467                :::*                    LISTEN      21222/java         
+ l, ~9 A: T" @" \( Itcp6       0      0 :::40571                :::*                    LISTEN      20780/java         

, u1 T! ]0 O, j8 u' ]