zookeeper运行之后可以使用 ./bin/zkCli.sh -server 127.0.0.1:2181 可以进入zookeeper的客户端,ls / 命令显示的zookeeper内部状态如下所示。
[zk: 127.0.0.1:2181(CONNECTED) 0] ls / [admin, brokers, cluster, config, consumers, controller, controller_epoch, isr_change_notification, zookeeper]上面的所有文件除了zookeeper文件夹是zookeeper产生的其他均为kafka所产生的。可以查看broker下面的相关信息。
[zk: 127.0.0.1:2181(CONNECTED) 12] ls /brokers [ids, seqid, topics] [zk: 127.0.0.1:2181(CONNECTED) 13] ls /brokers/ids [1, 2] [zk: 127.0.0.1:2181(CONNECTED) 14] ls /brokers/ids/1 [] [zk: 127.0.0.1:2181(CONNECTED) 15] get /brokers/ids/1 {"jmx_port":-1,"timestamp":"1494050585414","endpoints":["PLAINTEXT://192.168.128.128:19092"],"host":"192.168.128.128","version":3,"port":19092}上面显示了ids1的启动时间,端口号,版本
[zk: 127.0.0.1:2181(CONNECTED) 16] ls /brokers/topics [__consumer_offsets, my-first-kafka-topic, test] [zk: 127.0.0.1:2181(CONNECTED) 17] get /brokers/topics/test {"version":1,"partitions":{"0":[1]}}上面显示了当前存在的所有的topic。例如其中的test topic,只有一个分区 ,版本是1.
kafka的config文件夹下的,consumer.properties配置项如下:
eversilver@debian:/usr/local/kafka$ cat ./config/consumer.properties # Licensed to the Apache Software Foundation (ASF) under one or more # 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 # the License. You may obtain a copy of the License at # # # # 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. # See the License for the specific language governing permissions and # limitations under the License. # see kafka.consumer.ConsumerConfig for more details # Zookeeper connection string # comma separated host:port pairs, each corresponding to a zk # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" zookeeper.connect=127.0.0.1:2181 # timeout in ms for connecting to zookeeper zookeeper.connection.timeout.ms=6000 # consumer group id # 用来组合一个topic下面的多个partition,如果一个topic有两个group-id,那么 # 一个group-id对应于一个consumer组。不同consumer组可以复制消费这个topic的 # 消息。即一个topic消息可以被两个group拿两次。 group.id=test-consumer-group #consumer timeout #consumer.timeout.ms=5000kafka的config文件夹下的,producer.properties配置项也需要注意些。
注意,kafka的配置项优先级最高的是程序中设置的配置项,其次是shell中开始命令时的配置项,最低的是配置文件中设置的配置项。
相关资料:
: kafka快速开始
: broker配置项以及其他的配置项。
下面关于Kafka的文章您也可能喜欢,不妨参考下:
CentOS 7.2部署Elasticsearch+Kibana+Zookeeper+Kafka