kafka为分布式部署

kafka分布式部署：
就是在一台机器上安装3个zookeeper服务。
准备条件：
需要配置java环境变量：
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

1，解压安装包，并重命名为zookeeper01

2，修改配置文件
[kafka@kafka1 opt]$ cd zookeeper01/conf/
[kafka@kafka1 conf]$ cp zoo_sample.cfg zoo.cfg
修改zoo.cfg 如下：
[kafka@kafka1 conf]$ cat zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
#第一处主要修dataDir，我这里是zookeeper_data1，
dataDir=/opt/zookeeper_data/zookeeper_data1 
clientPort=2181
autopurge.purgeInterval=1
#修改交互的端口，因为我是都部署在一个节点上，所以需要写三个server，并且端口要做区分。
server.1=192.168.28.99:2888:3888
server.2=192.168.28.99:2887:3887
server.3=192.168.28.99:2886:3886

3：创建文件夹：
mkdir -p /opt/zookeeper_data/zookeeper_data1/logs
mkdir -p /opt/zookeeper_data/zookeeper_data2/logs
mkdir -p /opt/zookeeper_data/zookeeper_data3/logs

echo '1' > /opt/zookeeper_data/zookeeper_data1/myid
echo '2' > /opt/zookeeper_data/zookeeper_data2/myid
echo '3' > /opt/zookeeper_data/zookeeper_data3/myid

4：部署另外两个服务，并修改配置文件：
cp zookeeper01 zookeeper02
cp zookeeper01 zookeeper03

将zoo.cfg中的dataDir 做出相应修改
[kafka@kafka1 conf]$ cat /opt/zookeeper01/conf/zoo.cfg | grep dataDir
dataDir=/opt/zookeeper_data/zookeeper_data1
# The number of snapshots to retain in dataDir
[kafka@kafka1 conf]$ cat /opt/zookeeper02/conf/zoo.cfg | grep dataDir
dataDir=/opt/zookeeper_data/zookeeper_data2
# The number of snapshots to retain in dataDir
[kafka@kafka1 conf]$ cat /opt/zookeeper03/conf/zoo.cfg | grep dataDir
dataDir=/opt/zookeeper_data/zookeeper_data3
# The number of snapshots to retain in dataDir

将zoo.cfg中的clientPort 做出相应修改
[kafka@kafka1 conf]$ cat /opt/zookeeper03/conf/zoo.cfg | grep clientPort
clientPort=2183
[kafka@kafka1 conf]$ cat /opt/zookeeper02/conf/zoo.cfg | grep clientPort
clientPort=2182
[kafka@kafka1 conf]$ cat /opt/zookeeper01/conf/zoo.cfg | grep clientPort
clientPort=2181
[kafka@kafka1 conf]$

5，启动
[kafka@kafka1 opt]$ /opt/zookeeper01/bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper01/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[kafka@kafka1 opt]$ /opt/zookeeper02/bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper02/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[kafka@kafka1 opt]$ /opt/zookeeper03/bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper03/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[kafka@kafka1 opt]$

查看状态每个都有查看：
[kafka@kafka1 conf]$ /opt/zookeeper03/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper03/bin/../conf/zoo.cfg
Mode: follower
[kafka@kafka1 conf]$ /opt/zookeeper02/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper02/bin/../conf/zoo.cfg
Mode: leader
[kafka@kafka1 conf]$ /opt/zookeeper01/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper01/bin/../conf/zoo.cfg
Mode: follower
[kafka@kafka1 conf]$

6，jps验证，显示3个 QuorumPeerMain就是正常的
[kafka@kafka1 conf]$ jps
44385 QuorumPeerMain
44342 QuorumPeerMain
44313 QuorumPeerMain
44506 Jps
[kafka@kafka1 conf]$

KAFKA：
1，kafka参数配置如下：
[kafka@kafka1 opt]$ cat /opt/kafka/config/server.properties  | grep -ivE "#|^$"
broker.id=0
delete.topic.enable=true
#这里listeners需要注意
listeners=PLAINTEXT://192.168.28.99: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/kafka/logs
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
#还有zookeeper.connect需要注意
zookeeper.connect=192.168.28.99:2181,192.168.28.99:2182,192.168.28.99:2183
zookeeper.connection.timeout.ms=6000
log.retention.bytes=-1
[kafka@kafka1 opt]$

2，启动：
/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties>/opt/kafka/logs/kk.log &

验证
[kafka@kafka1 kafka]$ netstat -tunlp|egrep "(2181|9092)"
(Not all processes could be identified, non-owned process info
will not be shown, you would have to be root to see it all.)
tcp6       0      0 192.168.28.99:9092      :::*                    LISTEN      44545/java
tcp6       0      0 :::2181                 :::*                    LISTEN      44313/java
[kafka@kafka1 kafka]$

创建topic
我这里只有一个节点，所以没有副本--replication-factor 1
[kafka@kafka1 kafka]$ bin/kafka-topics.sh --create --zookeeper 192.168.28.99:2183 --replication-factor 1 --partitions 6 --topic t1
Created topic "t1".
[kafka@kafka1 kafka]$ bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --zookeeper 192.168.28.99:2183 --topic test1                       Created topic "test1".
[kafka@kafka1 kafka]$ bin/kafka-topics.sh --list --zookeeper 192.168.28.99:2181
t1
test1

同时打开两个session，在生产者端输入的内容，在消费者端可以显示出来

session1:
[kafka@kafka1 kafka]$ bin/kafka-console-producer.sh --broker-list 192.168.28.99:9092 --topic test1

hello my name is tanshuang
gcdw test 1

session2:
[kafka@kafka1 kafka]$ bin/kafka-console-consumer.sh --zookeeper 192.168.28.99:2181,192.168.28.99:2182,192.168.28.99:2183 --topic test1
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].

hello my name is tanshuang
gcdw test 1

[kafka@kafka1 kafka]$ bin/kafka-topics.sh --describe --zookeeper 192.168.28.99:2181 --topic t1
Topic:t1        PartitionCount:6        ReplicationFactor:1     Configs:
       Topic: t1       Partition: 0    Leader: 0       Replicas: 0     Isr: 0
       Topic: t1       Partition: 1    Leader: 0       Replicas: 0     Isr: 0
       Topic: t1       Partition: 2    Leader: 0       Replicas: 0     Isr: 0
       Topic: t1       Partition: 3    Leader: 0       Replicas: 0     Isr: 0
       Topic: t1       Partition: 4    Leader: 0       Replicas: 0     Isr: 0
       Topic: t1       Partition: 5    Leader: 0       Replicas: 0     Isr: 0
[kafka@kafka1 kafka]$

完成