./bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --list

./bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group kafkademo_test_group

./bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --time -1 --topic kafka_demo_test_topic
kafka_demo_test_topic:2:169043
kafka_demo_test_topic:5:169039
kafka_demo_test_topic:4:169036
kafka_demo_test_topic:1:169043
kafka_demo_test_topic:3:169039
kafka_demo_test_topic:0:169043


发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka_demo_test_topic   
>a
>b
>c


消费消息
dugang$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka_demo_test_topic 
a
b
c




性能压测

./kafka-producer-perf-test.sh --topic kafka_demo_test_topic --num-records 1000000 --throughput -1 --record-size 128 --producer-props bootstrap.servers=localhost:9092 acks=-1 linger.ms=2000 compression.type=lz4

./kafka-consumer-perf-test.sh --broker-list localhost:9092 --messages 1000000 --topic producer


0.8以前的，offset 保存在zk中，consumer需要配置zk地址
之后版本由broker管理，指定 bootstrap-server 即可

bootstrap-server 可以自动发现其它的broker ，写一个就行


Java 客户端 生产者和消费者 使用 bootstrap.servers


Properties props = new Properties();
props.put("bootstrap.servers", KafkaUtil.SERVERS);
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<String, String>(props);