• 【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
  • 相关阅读:
    定时任务框架
    记一次前端CryptoJS AES解密
    35-SpringBoot 短信验证码接口使用
    UnityShader 全局传值无效
    比尔·盖茨晒48年前简历:“没你们的好看”
    Day 1 Vue 页面框架
    sort方法实现复杂排序
    Android Camera性能分析 - 第25讲 CameraServer LatencyHistogram简介
    sklearn机器学习——day14
    [杂谈]-快速了解直接内存访问 (DMA)
  • 原文地址:https://blog.csdn.net/qq_42251120/article/details/132788106