Kafka简单Java示例

import java.util.Properties;import org.apache.kafka.clients.producer.Producer;import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.clients.producer.ProducerRecord;public class SimpleProducer { public static void main(String[] args) throws Exception{String topicName = "Hello-Kafka";Properties props = new Properties();// 如果使用IP地址连接需要在kafka配置文件service.properties// advertised.listeners=PLAINTEXT://YOURSELF-IP:9092props.put("bootstrap.servers", "localhost:9092");// Set acknowledgements for producer requests、 props.put("acks", "all");// If the request fails, the producer can automatically retry,props.put("retries", 0);//Specify buffer size in configprops.put("batch.size", 16384);// Reduce the no of requests less than 0 props.put("linger.ms", 1);// The buffer.memory controls the total amount // of memory available to the producer for buffering、 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 producer = new KafkaProducer<>(props);for(int i = 0; i < 10; i++) {String key = Integer.toString(i);String value = Integer.toString(i);producer.send(new ProducerRecord<>(topicName, key, value));}System.out.println("Message sent successfully");producer.close(); }}

import java.util.Properties;import java.util.Arrays;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.ConsumerRecord;public class SimpleConsumer { public static void main(String[] args) throws Exception { String topicName = "Hello-Kafka"; Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test"); props.put("enable.auto.commit", "true"); 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 consumer = new KafkaConsumer<>(props); // Kafka Consumer subscribes list of topics here. consumer.subscribe(Arrays.asList(topicName)) //print the topic name System.out.println("Subscribed to topic: " + topicName); while (true) { ConsumerRecords records = consumer.poll(100); for (ConsumerRecord record : records) // print the offset,key and value for the consumer records. System.out.printf("offset = %d, key = %s, value = %sn", record.offset(), record.key(), record.value()); } }}