Java實現Kafka生產者和消費者的示例


 

 

 
Java實現Kafka生產者和消費者的示例

Kafka簡介

Kafka是由Apache軟件基金會開發的一個開源流處理平台,由Scala和Java編寫。Kafka的目標是為處理實時數據提供一個統一、高吞吐、低延遲的平台。

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

方式一:kafka-clients

引入依賴

在pom.xml文件中,引入kafka-clients依賴:

<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.3.1</version> </dependency> 

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

生產者

創建一個KafkaProducer的生產者實例:

@Configuration public class Config { public final static String bootstrapServers = "127.0.0.1:9092"; @Bean(destroyMethod = "close") public KafkaProducer<String, String> kafkaProducer() { Properties props = new Properties(); //設置Kafka服務器地址 props.put("bootstrap.servers", bootstrapServers); //設置數據key的序列化處理類 props.put("key.serializer", StringSerializer.class.getName()); //設置數據value的序列化處理類 props.put("value.serializer", StringSerializer.class.getName()); KafkaProducer<String, String> producer = new KafkaProducer<>(props); return producer; } } 

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

在Controller中進行使用:

@RestController @Slf4j public class Controller { @Autowired private KafkaProducer<String, String> kafkaProducer; @RequestMapping("/kafkaClientsSend") public String send() { String uuid = UUID.randomUUID().toString(); RecordMetadata recordMetadata = null; try { //將消息發送到Kafka服務器的名稱為“one-more-topic”的Topic中 recordMetadata = kafkaProducer.send(new ProducerRecord<>("one-more-topic", uuid)).get(); log.info("recordMetadata: {}", recordMetadata); log.info("uuid: {}", uuid); } catch (Exception e) { log.error("send fail, uuid: {}", uuid, e); } return uuid; } } 

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

消費者

創建一個KafkaConsumer的消費者實例:

@Configuration public class Config { public final static String groupId = "kafka-clients-group"; public final static String bootstrapServers = "127.0.0.1:9092"; @Bean(destroyMethod = "close") public KafkaConsumer<String, String> kafkaConsumer() { Properties props = new Properties(); //設置Kafka服務器地址 props.put("bootstrap.servers", bootstrapServers); //設置消費組 props.put("group.id", groupId); //設置數據key的反序列化處理類 props.put("key.deserializer", StringDeserializer.class.getName()); //設置數據value的反序列化處理類 props.put("value.deserializer", StringDeserializer.class.getName()); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props); //訂閱名稱為“one-more-topic”的Topic的消息 kafkaConsumer.subscribe(Arrays.asList("one-more-topic")); return kafkaConsumer; } } 

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

在Controller中進行使用:

@RestController @Slf4j public class Controller { @Autowired private KafkaConsumer<String, String> kafkaConsumer; @RequestMapping("/receive") public List<String> receive() { ////從Kafka服務器中的名稱為“one-more-topic”的Topic中消費消息 ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(1)); List<String> messages = new ArrayList<>(records.count()); for (ConsumerRecord<String, String> record : records.records("one-more-topic")) { String message = record.value(); log.info("message: {}", message); messages.add(message); } return messages; } } 

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

方式二:spring-kafka

使用kafka-clients需要我們自己創建生產者或者消費者的bean,如果我們的項目基於SpringBoot構建,那么使用spring-kafka就方便多了。

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

引入依賴

在pom.xml文件中,引入spring-kafka依賴:

<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>2.3.12.RELEASE</version> </dependency> 

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

生產者

在application.yml文件中增加配置:

spring: kafka: #Kafka服務器地址 bootstrap-servers: 127.0.0.1:9092 producer: #設置數據value的序列化處理類 value-serializer: org.apache.kafka.common.serialization.StringSerializer 

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

在Controller中注入KafkaTemplate就可以直接使用了,代碼如下:

@RestController @Slf4j public class Controller { @Autowired private KafkaTemplate<String, String> template; @RequestMapping("/springKafkaSend") public String send() { String uuid = UUID.randomUUID().toString(); //將消息發送到Kafka服務器的名稱為“one-more-topic”的Topic中 this.template.send("one-more-topic", uuid); log.info("uuid: {}", uuid); return uuid; } } 

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

消費者

在application.yml文件中增加配置:

spring: kafka: #Kafka服務器地址 bootstrap-servers: 127.0.0.1:9092 consumer: #設置數據value的反序列化處理類 value-deserializer: org.apache.kafka.common.serialization.StringDeserializer 

文章持續更新,微信搜索「萬貓學社」第一時間閱讀,關注后回復「電子書」,免費獲取12本Java必讀技術書籍。

創建一個可以被Spring框架掃描到的類,並且在方法上加上@KafkaListener注解,就可以消費消息了,代碼如下:

@Component @Slf4j public class Receiver { @KafkaListener(topics = "one-more-topic", groupId = "spring-kafka-group") public void listen(ConsumerRecord<?, ?> record) { Optional<?> kafkaMessage = Optional.ofNullable(record.value()); if (kafkaMessage.isPresent()) { String message = (String) kafkaMessage.get(); log.info("message: {}", message); } } }


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM