kafka集群搭建

####安装kafka

# tar xzvf kafka_2.9.1-0.8.2.1.tgz

# cd kafka_2.9.1-0.8.2.1

# vi config/server.properties

broker.id=0       #每一个kafka的id都不同
port=9092
host.name=192.168.10.91    #监听的接口ip
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=/data/app/kafka_2.9.1-0.8.2.1/log
num.partitions=1         #默认建立的partition数量为1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect=192.168.10.91:2181,192.168.10.92:2181,192.168.10.93:2181      #3个节点的zookeeper集群
zookeeper.connection.timeout.ms=6000

# mkdir log

# ./bin/kafka-server-start.sh -daemon /data/app/kafka_2.9.1-0.8.2.1/config/server.properties #启动kafka

####测试kafka

• 建立topic

# ./bin/kafka-topics.sh --create --topic topic1 --zookeeper 192.168.10.91:2181 --partitions 3 --replication-factor 1

Created topic "topic1".

• 查看topic

# ./bin/kafka-topics.sh --topic topic1 --zookeeper 192.168.10.91:2181 --describe

Topic:topic1    PartitionCount:3        ReplicationFactor:1     Configs:
        Topic: topic1   Partition: 0    Leader: 0       Replicas: 0     Isr: 0
        Topic: topic1   Partition: 1    Leader: 0       Replicas: 0     Isr: 0
        Topic: topic1   Partition: 2    Leader: 0       Replicas: 0     Isr: 0

• 三台服务器上分别开启个comsumer # ./bin/kafka-console-consumer.sh --topic topic1 --zookeeper 192.168.10.91:2181,192.168.10.92:2181,192.168.10.93:2181

• 建立个productor，并发送消息 # ./bin/kafka-console-producer.sh --broker-list 192.168.10.91:9092,192.168.10.92:9092,192.168.10.93:9092 --topic pic1

[2016-04-09 01:25:05,447] WARN Property topic is not valid (kafka.utils.VerifiableProperties)
fsr   #随便输入点啥
fsrg

• 查看comsumer是否消费 # ./bin/kafka-console-consumer.sh --topic topic1 --zookeeper 192.168.10.91:2181,192.168.10.92:2181,192.168.10.93:2181

fsr
fsrg

查看kafkaoffsetmonitor