zoukankan      html  css  js  c++  java
  • Spring Boot Kafka

    Kafka的核心概念,以及如何保证高可用性、可靠性、顺序性等问题这里就不赘述了,本文只记录操作示例。

    依赖配置

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    

    application.yml

    spring:
      kafka:
        bootstrap-servers: localhost:9092
        consumer:
          group-id: message # consumer group 消费者群组
          auto-offset-reset: latest # 当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
    

    KafkaProducerConfig

    存储Java类MessageForm对象,需要自行替换

    @Configuration
    public class KafkaProducerConfig {
    
        @Value("${spring.kafka.bootstrap-servers}")
        private String bootstrapServers;
        
    
        @Bean
        public Map<String, Object> producerConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    
            return props;
        }
    
        @Bean
        public ProducerFactory<String, MessageForm> producerFactory() {
            return new DefaultKafkaProducerFactory<>(producerConfigs());
        }
    
        @Bean
        public KafkaTemplate<String, MessageForm> kafkaTemplate() {
            return new KafkaTemplate<>(producerFactory());
        }
    }
    

    KafkaConsumerConfig

    配置类上需要@EnableKafka注释才能在Spring托管Bean上检测@KafkaListener注解。

    @EnableKafka
    @Configuration
    public class KafkaConsumerConfig {
        @Value("${spring.kafka.bootstrap-servers}")
        private String bootstrapServers;
    
        @Bean
        public Map<String, Object> consumerConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    
            return props;
        }
    
        @Bean
        public ConsumerFactory<String, MessageForm> consumerFactory() {
            return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
                    new JsonDeserializer<>(MessageForm.class));
        }
    
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, MessageForm> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, MessageForm> factory =
                    new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
    
            return factory;
        }
    }
    

    KafkaMessageListener

    在KafkaConsumerConfig中我们配置了ConsumerFactory和KafkaListenerContainerFactory。当这两个Bean成功注册到Spring IOC容器中后,我们便可以使用@KafkaListener注解来监听消息了。

    @Component
    public class KafkaMessageListener {
    
        @Resource
        private MessageService messageServiceImpl;
    
        @KafkaListener(topics="message",groupId = "consumer")
        public void listen(MessageForm messageForm){
            messageServiceImpl.message(messageForm);
        }
    }
    
  • 相关阅读:
    Unity跳转场景进度条制作教程(异步加载)
    Unity LOD-Level of Detail(多层次细节)用法教程
    Unity查找物体的子物体、孙物体
    Unity导航系统Navigation使用教程
    Unity物品栏、商城3D物品的显示插件
    VIM编辑器常用命令
    linux系统常用命令
    CRITICAL:yum.cli:Config Error: Error accessing file for config file:///etc/yum.conf
    linux下使用tar命令详解
    Linux之安装软件
  • 原文地址:https://www.cnblogs.com/zenan/p/10893640.html
Copyright © 2011-2022 走看看