• 【Java Web】Spring整合Kafka


    • 引入依赖
      • spring-kafka
    • 配置Kafka
      • 配置server/consumer
    • 访问Kafka
      • 生产者
        • kafkaTemplate.send(topic,data)
      • 消费者
        • @KafkaListenner(topics={“test”})
        • public void handleMessage(ConsumerRecord record) { }

    application.properties配置:

    # kafka
    spring.kafka.bootstrap-servers=localhost:9092
    spring.kafka.consumer.group-id=community-consumer-group
    spring.kafka.consumer.enable-auto-commit=true
    spring.kafka.consumer.auto-commit-interval=3000
    
    • 1
    • 2
    • 3
    • 4
    • 5

    测试代码:

    package com.nowcoder.community;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Component;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    @ContextConfiguration(classes = CommunityApplication.class)
    public class KafkaTests {
    
        @Autowired
        private KafkaProducer kafkaProducer;
        @Test
        public void testKafka(){
            kafkaProducer.sendMessage("test","你好");
            kafkaProducer.sendMessage("test","兄弟");
    
            try {
                Thread.sleep(1000*10);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
    
    @Component
    class KafkaProducer{
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        public void sendMessage(String topic, String content){
            kafkaTemplate.send(topic, content);
        }
    }
    
    @Component
    class KafkaConsumer{
    
        @KafkaListener(topics = {"test"})
        public void handleMessage(ConsumerRecord record){
            System.out.println(record.value());
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
  • 相关阅读:
    Docker 常用命令
    软件测试面试题:性能测试工作?
    SpringCloudAlibaba—Nacos config
    最小体力消耗-分支限界法
    树状数组&线段树 的奇妙用法
    为什么选择事件驱动的微服务架构?
    工作总结:kafka踩过的坑
    MySQL批量入库的几种方式详解
    深度解读ChatGPT基本原理
    P35 set 找不同
  • 原文地址:https://blog.csdn.net/qq_42251120/article/details/132788106