kafka demo最简单的实现
包的结构
pom.xml相关的依赖:
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.0.0</version>
</dependency>
</dependencies>
Producer类:
package com;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
/**
* @author Modesty.P.Gao
* @version 1.0
* @description: TODO
* @date 11/11/2021 3:25 PM
*/
public class Producer {
private static final String TOPIC="demo";
private static final String BPOKER_LIST="localhost:9092";
private static KafkaProducer<String,String> producer=null;
static {
Properties configs=initConfig();
producer=new KafkaProducer<String, String>(configs);
}
private static Properties initConfig() {
Properties properties=new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BPOKER_LIST);
properties.put(ProducerConfig.ACKS_CONFIG,"all");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
return properties;
}
public static void main(String[] args) {
final String message="hello world Modesty";
ProducerRecord<String,String> record =new ProducerRecord<String, String>(TOPIC,message);
try {
producer.send(record, new Callback() {
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (null==exception){
System.out.println("perfect");
}
if (null!=metadata){
System.out.println("offset:"+metadata.offset()+";partition:"+metadata.partition());
}
}
}).get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}finally {
producer.close();
}
}
}
Consumer类
package com;
import org.apache.kafka.clients.consumer.ConsumerConfig;
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.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Arrays;
import java.util.Properties;
/**
* @author Modesty.P.Gao
* @version 1.0
* @description: TODO
* @date 11/11/2021 4:02 PM
*/
public class Consumer {
private static final String TOPIC="demo";
private static final String BPOKER_LIST="localhost:9092";
private static KafkaConsumer<String, String> consumer = null;
static {
Properties properties = initConfig();
consumer = new KafkaConsumer<String, String>(properties);
consumer.subscribe(Arrays.asList(TOPIC));
}
private static Properties initConfig() {
Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,BPOKER_LIST);
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
properties.put(ConsumerConfig.CLIENT_ID_CONFIG,"test");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
return properties;
}
public static void main(String[] args) {
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
try {
System.out.println(record.value());
} catch (Exception e) {
e.printStackTrace();
}
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
consumer.close();
}
}
}
运行结果