1.概要

继上文讲解了Kafka在CentOS环境中的安装和Kafka在CentsOS环境下单机集群的搭建，本文将基于Springboot环境，搭建生产者与消费者案例，利用KafkaProducer向kafka内发送消息，利用KafkaConsumer来获取消息并进行打印，直观展示消息的生产与消费。

2.生产者部分代码

1.生产者配置类

import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;import java.util.HashMap;
import java.util.Map;/*** @description: Kafka配置类* @author: marin* @time: 2020/8/3 19:02*/
@Configuration
@EnableKafka
public class KafkaProducerConfig {@Value("${spring.kafka.producer.bootstrap-servers}")private String servers;@Value("${spring.kafka.producer.retries}")private int retries;@Value("${spring.kafka.producer.batch.size}")private int batchSize;@Value("${spring.kafka.producer.linger}")private int linger;@Value("${spring.kafka.producer.buffer.memory}")private int bufferMemory;@Value("${spring.kafka.producer.key-serializer}")private String keyDeserializer;@Value("${spring.kafka.producer.value-serializer}")private String valueDeserializer;public Map<String, Object> producerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);props.put(ProducerConfig.RETRIES_CONFIG, retries);props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);props.put(ProducerConfig.LINGER_MS_CONFIG, linger);props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keyDeserializer);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueDeserializer);return props;}public ProducerFactory<String, String> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigs());}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<String, String>(producerFactory());}
}

2.生产者配置

spring.kafka.producer.bootstrap-servers=CentOSIP:9092 //你Kafka的ip地址
spring.kafka.producer.retries=0
spring.kafka.producer.batch.size=4096
spring.kafka.producer.linger=1
spring.kafka.producer.buffer.memory=40960
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

3.消费者部分代码

1.消费者配置类

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;import java.util.HashMap;
import java.util.Map;/*** @description: Kafka消费配置类* @author: marin* @time: 2020/8/3 20:02*/@Configuration
public class KafkaConsumerConfig {@Value("${spring.kafka.consumer.bootstrap-servers}")private String bootstrapServers;@Value("${spring.kafka.consumer.enable-auto-commit}")private String enableAutoCommit;@Value("${spring.kafka.consumer.auto-offset-reset}")private String autoOffsetReset;@Value("${spring.kafka.consumer.auto-commit-interval}")private String autoCommitInterval;@Value("${spring.kafka.consumer.key-deserializer}")private String keyDeserializer;@Value("${spring.kafka.consumer.value-deserializer}")private String valueDeserializer;@Value("${spring.kafka.consumer.group-id}")private String groupId;public Map<String, Object> getCommonProperties() {Map<String, Object> properties = new HashMap<String, Object>();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);/**默认 group id */properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);return properties;}/*** kafka bean 容器*/@Beanpublic KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<String, String>();factory.setConsumerFactory(new DefaultKafkaConsumerFactory<String, String>(getCommonProperties()));return factory;}
}

2.消费者部分配置

spring.kafka.consumer.bootstrap-servers=CentOSIP:9092 //你Kafka的ip地址
spring.kafka.consumer.group-id=test
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.auto-commit-interval=1000
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
kafka.consumer.topic=demo

3.消费者消费方法

import com.awifi.demo.conf.KafkaConsumerConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.apache.kafka.clients.consumer.KafkaConsumer;import java.time.Duration;
import java.util.Collections;/*** @description: ToDo* @author: marin* @time: 2020/8/7 15:52*/
@Slf4j
@Component
public class KafkaConsumerExample {@Autowiredprivate KafkaConsumerConfig kafkaConsumerConfig;@Value("${kafka.consumer.topic}")private String topic;public void run() {KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafkaConsumerConfig.getCommonProperties());// 订阅test1 topicconsumer.subscribe(Collections.singletonList(topic));while(true) {//  从服务器开始拉取数据ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));records.forEach(record -> {log.error("topic:{},key:{},value:{}",record.topic(),record.key(),record.value());});}}
}

4.Controller编写

import com.awifi.demo.consumer.KafkaConsumerExample;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;/*** @description: ToDo* @author: marin* @time: 2020/8/7 15:13*/
@Slf4j
@RestController
@RequestMapping("/kafka")
public class KafkaController {@Autowiredprivate KafkaTemplate kafkaTemplate;@Autowiredprivate KafkaConsumerExample kafkaConsumer;@PostMapping("/test")public void kafkaTest(String message) {kafkaTemplate.send("demo",message);}@PostMapping("/consumer")public boolean kafkaConsumer() {kafkaConsumer.run();return true;}
}

5.测试结果

1.发送消息

2.消费消息

3.控制台输出结果

每发送一次消息，控制台都会打印输出结果。

6.小结

本文演示了一个简单的Kafka生产消息与消费消息的案例，希望给新手入门Kafka一点启示，其中的pom文件没有详细列出来，需要的可以查看下面的git地址获取，下篇将讲述Kafka多线程消费与配置的一些原理。

7.本文代码地址

码云：https://gitee.com/Marinc/kafkademo