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

引入依賴

在pom.xml文件中，引入kafka-clients依賴：

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

生產者

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

@Configurationpublic 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; }}

在Controller中進行使用：

@RestController@Slf4jpublic 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; }}

消費者

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

@Configurationpublic 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; }}

在Controller中進行使用：

@RestController@Slf4jpublic 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; }}方式二：spring-kafka

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

引入依賴

在pom.xml文件中，引入spring-kafka依賴：

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

生產者

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

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

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

@RestController@Slf4jpublic 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; }}

消費者

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

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

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

@Component@Slf4jpublic 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); } }}

到此這篇關于Java實現Kafka生產者和消費者的示例的文章就介紹到這了,更多相關Java Kafka生產者和消費者 內容請搜索好吧啦網以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持好吧啦網！