Kafka 简单Java示例

不念不忘少年蓝@ 2023-10-01 19:52 35阅读 0赞

# `一、生产者` #

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:9092
    		props.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 config
    		props.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<String, String> 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<String, String> 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<String, String> records = consumer.poll(100);
             for (ConsumerRecord<String, String> record : records)
             // print the offset,key and value for the consumer records.
             System.out.printf("offset = %d, key = %s, value = %s\n", 
             record.offset(), record.key(), record.value());
          }
       }
    }