kafka学习 -- 客户端Consumer API学习

以你之姓@ 2023-02-22 05:16 57阅读 0赞

## 写在前面 ##

这里我会总结一下，Springboot 集成 spring-kafka中，consumer 的相关配置，Api

这里的东西，比 Producer 稍微多一些

集成相关配置

server:
      port: 9000
    spring:
      kafka:
        bootstrap-servers: 192.168.1.74:9092
        consumer:
          group-id: group_id
          # 手动提交
          enable-auto-commit: false
          auto-offset-reset: earliest
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          properties:
            session.timeout.ms: 60000
        listener:
          type: batch
          log-container-config: false
          concurrency: 3
          # 手动提交
          ack-mode: manual_immediate

## 一、@KafkaListener ##

Spring Kafka 集成Kafka，消息的监听者配置，使用特别简单，只有一个注解即可实现监听

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

虽然只有一个注解配置，需要注意消息的序列化，以及相关监听处理

## 二、代码示例 ##

示例一

@KafkaListener(topics = "users", groupId = "group_id")
        public void consume(String message, Acknowledgment acknowledgment) throws IOException { 
            try { 
                logger.info(String.format("#### -> Consumed message -> %s", message));
            } catch (Exception e) { 
                logger.error(e.getMessage(), e);
            } finally { 
                // 手动提交 offset
                acknowledgment.acknowledge();
            }
        }

示例二

import com.common.Bar2;
    import com.common.Foo2;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.annotation.KafkaHandler;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    /** * 这里定义了消费者组，和多个 topic 的对应 * * @author Gary Russell * @since 5.1 */
    @Component
    @KafkaListener(id = "multiGroup", topics = { "foos", "bars", "test"})
    public class MultiMethods { 
    
        private final Logger logger = LoggerFactory.getLogger(MultiMethods.class);
    
        @KafkaHandler
        public void foo(Foo2 foo) { 
            System.out.println("Received: " + foo);
        }
    
        @KafkaHandler
        public void bar(Bar2 bar) { 
            System.out.println("Received: " + bar);
        }
    
        @KafkaHandler(isDefault = true)
        public void unknown(Object object) { 
            System.out.println("Received unknown: " + object);
        }
    
    
        @KafkaHandler
        public void tests(ConsumerRecord record) { 
            logger.info("Received -> key :{} ", record.key());
        }
    
    }

## 三、更多Kafka,，客户端、服务端，监控可参考下文 ##

*  [kafka学习 – kafka connect][kafka_ _ kafka connect]
 *  [kafka学习 – 客户端Consumer API学习][kafka_ _ _Consumer API]
 *  [kafka学习 – 客户端Producer API学习][kafka_ _ _Producer API]
 *  [kafka学习 – 服务端命令学习][kafka_ _]
 *  [Springboot集成Kafka][Springboot_Kafka]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200705195756879.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyMTA1NjI5,size_16,color_FFFFFF,t_70
[kafka_ _ kafka connect]: https://blog.csdn.net/qq_42105629/article/details/107077881
[kafka_ _ _Consumer API]: https://blog.csdn.net/qq_42105629/article/details/107073062
[kafka_ _ _Producer API]: https://blog.csdn.net/qq_42105629/article/details/107072983
[kafka_ _]: https://blog.csdn.net/qq_42105629/article/details/107008616
[Springboot_Kafka]: https://blog.csdn.net/qq_42105629/article/details/100569364