客戶端命令: 訊息者（全） kafka-console-consumer.sh --zookeeper localhost:2181 --topic my-topic--from-beginning 生產者     bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-topic JAVA生產、消費(直接上官方例子) 生產都 import org.apache.kafka.clients.producer.*; import java.util.Properties; public class KafKaProducerTest { public static void main(String[] args) { Properties props = new Properties(); //kafka broker的地址 props.put("bootstrap.servers", "172.21.61.42:9092,172.21.61.43:9092,172.21.61.44:9092"); //ack配置，預設為0，0：producer不會等待broker傳送ack，1：當leader接收到訊息之後傳送ack，all：當所有的follower都同步訊息成功後傳送ack props.put("acks", "all"); //key的序列化方式 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //value的序列化方式 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //生產者執行緒安全，需要放到單例中，初始化一次 Producer<String, String> producer = new KafkaProducer<String, String>(props); for(int i = 0; i < 100; i++) producer.send(new ProducerRecord<String, String>("news_gather_topic", "message: " + i++)); producer.close(); System.out.println("done"); } } 消費者 import java.util.Arrays; import java.util.List; import java.util.Properties; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.TopicPartition; public class KafkaConsumerTest { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "172.21.61.42:9092,172.21.61.43:9092,172.21.61.44:9092"); props.put("group.id", "test");//不同ID 可以同時訂閱訊息 props.put("enable.auto.commit", "false"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); consumer.subscribe(Arrays.asList("test", "atcotest" , "news_gather_topic"));//訂閱TOPIC try { while(true) {//輪詢 ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE); for (TopicPartition partition : records.partitions()) { List<ConsumerRecord<String, String>> partitionRecords = records.records(partition); for (ConsumerRecord<String, String> record : partitionRecords) { //可以自定義Handler,處理對應的TOPIC訊息(partitionRecords.key()) System.out.println(record.offset() + ": " + record.value()); } consumer.commitSync();//同步 } } } finally { consumer.close(); } } }