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

写在前面

这里我会总结一下，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，消息的监听者配置，使用特别简单，只有一个注解即可实现监听

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

二、代码示例

示例一

@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 connectkafka学习 – 客户端Consumer API学习kafka学习 – 客户端Producer API学习kafka学习 – 服务端命令学习Springboot集成Kafka