Kafka是由Apache軟件基金會(huì)開發(fā)的一個(gè)開源流處理平臺(tái)，由Scala和Java編寫。Kafka的目標(biāo)是為處理實(shí)時(shí)數(shù)據(jù)提供一個(gè)統(tǒng)一、高吞吐、低延遲的平臺(tái)。

引入依賴

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

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

生產(chǎn)者

創(chuàng)建一個(gè)KafkaProducer的生產(chǎn)者實(shí)例：

@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(); //設(shè)置Kafka服務(wù)器地址 props.put('bootstrap.servers', bootstrapServers); //設(shè)置數(shù)據(jù)key的序列化處理類 props.put('key.serializer', StringSerializer.class.getName()); //設(shè)置數(shù)據(jù)value的序列化處理類 props.put('value.serializer', StringSerializer.class.getName()); KafkaProducer<String, String> producer = new KafkaProducer<>(props); return producer; }}

在Controller中進(jìn)行使用：

@RestController@Slf4jpublic class Controller { @Autowired private KafkaProducer<String, String> kafkaProducer; @RequestMapping('/kafkaClientsSend') public String send() { String uuid = UUID.randomUUID().toString(); RecordMetadata recordMetadata = null; try { //將消息發(fā)送到Kafka服務(wù)器的名稱為“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; }}

消費(fèi)者

創(chuàng)建一個(gè)KafkaConsumer的消費(fèi)者實(shí)例：

@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(); //設(shè)置Kafka服務(wù)器地址 props.put('bootstrap.servers', bootstrapServers); //設(shè)置消費(fèi)組 props.put('group.id', groupId); //設(shè)置數(shù)據(jù)key的反序列化處理類 props.put('key.deserializer', StringDeserializer.class.getName()); //設(shè)置數(shù)據(jù)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中進(jìn)行使用：

@RestController@Slf4jpublic class Controller { @Autowired private KafkaConsumer<String, String> kafkaConsumer; @RequestMapping('/receive') public List<String> receive() { 從Kafka服務(wù)器中的名稱為“one-more-topic”的Topic中消費(fèi)消息 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需要我們自己創(chuàng)建生產(chǎn)者或者消費(fèi)者的bean，如果我們的項(xiàng)目基于SpringBoot構(gòu)建，那么使用spring-kafka就方便多了。

引入依賴

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

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

生產(chǎn)者

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

spring: kafka: #Kafka服務(wù)器地址 bootstrap-servers: 127.0.0.1:9092 producer: #設(shè)置數(shù)據(jù)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(); //將消息發(fā)送到Kafka服務(wù)器的名稱為“one-more-topic”的Topic中 this.template.send('one-more-topic', uuid); log.info('uuid: {}', uuid); return uuid; }}

消費(fèi)者

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

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

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

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

到此這篇關(guān)于Java實(shí)現(xiàn)Kafka生產(chǎn)者和消費(fèi)者的示例的文章就介紹到這了,更多相關(guān)Java Kafka生產(chǎn)者和消費(fèi)者 內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！