【Kafka】基本使用：SpringBoot整合Kafka-clinets

迷南。 2022-12-21 04:48 133阅读 0赞

首先说明一点，SpringBoot 的版本和 Kafka-clinets 的版本有一个对照表格，如果没有按照正确的版本来引入，那么会存在版本问题导致ClassNotFound的问题，具体请参考 [https://spring.io/projects/spring-kafka][https_spring.io_projects_spring-kafka]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]

依赖：

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

## 1.Producer ##

@Component
    public class MyKafKaProducer {
        
    
        @Resource
        // Integer：消息key的类型，每条消息都需一个 key，它决定了该消息会去哪个 Partition
        // String：消息类型
        private KafkaTemplate<Integer, String> kafkaTemplate;  
    
        public void send() {
        
        	// 通过kafka进行消息发送
        	// ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
            kafkaTemplate.send("test", 1, "msgData");
        }
    }

## 2.Consumer ##

@Component
    public class MyKafkaConsumer {
        
    	
        // 订阅相应主题。接收到的消息会被注入ConsumerRecord 
        @KafkaListener(topics = {
        "test"})  
        public void listener(ConsumerRecord record) {
        
        
            // Optional，jdk8新特性，用于null处理
            Optional msg = Optional.ofNullable(record.value()); 
            // isPresent判斷值是否存在
            if (msg.isPresent()) {
          
                // 打印消息          
                System.out.println(msg.get()); // get，从optional中取出对应值
            }
        }
    }

## 3.application.properties ##

# kafka服务器
    spring.kafka.bootstrapservers=43.107.136.126:9092
    # 编码（Producer）
    spring.kafka.producer.keyserializer=org.apache.kafka.common.serialization.IntegerSerializer 
    spring.kafka.producer.valueserializer=org.apache.kafka.common.serialization.StringSerializer
    # 解码（Consumer）
    spring.kafka.consumer.keydeserializer=org.apache.kafka.common.serialization.IntegerDeserializer
    spring.kafka.consumer.valuedeserializer=org.apache.kafka.common.serialization.StringDeserializer
    # 对Consumer的特殊配置
    spring.kafka.consumer.group-id=test-consumer-group 
    spring.kafka.consumer.auto-offset-reset=earliest 
    spring.kafka.consumer.enable-auto-commit=true

> 注：如果我们关闭自动提交，即 enable-auto-commit 为 false 时，我们还可以配置 spring.kafka.listener.concurrency 去配置监听消息的线程数，即每个 @KafkaListener 注解标识的 topic 会有几个线程从队列中取出消息。具体参考请[这篇文章][Link 1]…

## 4.启动类 ##

@SpringBootApplication
    public class TestKafkaApplication {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            ConfigurableApplicationContext context = 
                SpringApplication.run(TestKafkaApplication.class, args);
                
            // 获取producer的Bean实例，send消息
            MyKafKaProducer kafKaProducer = context.getBean(MyKafKaProducer.class);
            for (int i = 0; i < 10; i++) {
        
            	// Producer发送消息
            	// 发送之后Consumer自己就会poll，然后打印
                kafKaProducer.send();
                TimeUnit.SECONDS.sleep(2);
            }
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1][]

[https_spring.io_projects_spring-kafka]: https://spring.io/projects/spring-kafka
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/9d45bc613e794aa2aeeec88b4b8779ff.png
[Link 1]: https://www.jianshu.com/p/ad0e5424edbd
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/2020111420102460.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw==,size_16,color_FFFFFF,t_70#pic_center