2018-07-18 kafka

分布式消息队列 分为生产者和消费者
scala写的

集群部署：
1.安装zookeeper
2.解压kafka
3.修改配置文件 vi /conf/server.properties

#id，每台机器id不一致 可用0，1，2
broker.id=0
#端口号
prot=9092
#host每台机器对用自己的ip
host.name=ip1
#与上方保持一致
advertised.host.name=ip1
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=/usr/local/myfile/kafka/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
#注：此处/kafka是将kafka数据全部存入zk中/kafka下，所以下方连接shell操作时也必须用zkip:2181/kafka才能正确操作
zookeeper.connect=hadoop1:2181,hadoop2:2181,hadoop3:2181/kafka
zookeeper.connection.timeout.ms=6000

4.启动
每台机器输入

bin/kafka-server-start.sh  config/server.properties

5.shell操作（如果上方配置在zk/kafka下，下方的连接zk全部要加/kafka）
查看当前服务器中的所有topic
bin/kafka-topics.sh --list --zookeeper zk01:2181（/kafka）
创建topic
bin/kafka-topics.sh --create --zookeeper zk01:2181 --replication-factor 1 --partitions 1 --topic test
删除topic
sh bin/kafka-topics.sh --delete --zookeeper zk0 --topic test
需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启。
通过shell命令发送消息
kafka-console-producer.sh --broker-list kafka01:9092 --topic itheima
通过shell消费消息
sh bin/kafka-console-consumer.sh --zookeeper zk01:2181 --from-beginning --topic test1
查看消费位置
sh kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper zk01:2181 --group testGroup
查看某个Topic的详情
sh kafka-topics.sh --topic test --describe --zookeeper zk01:2181
对分区数进行修改
kafka-topics.sh --zookeeper zk01 --alter --partitions 15 --topic utopic