kafka教程-springboot实现简单消费者

作者：马育民 • • 阅读：10353

创建springboot工程

https://www.malaoshi.top/show_1EF5qXjMmlZR.html

添加依赖

spring框架有对kafka的支持，添加该依赖

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.4.12.RELEASE</version>
</dependency>

选择 2.4.12.RELEASE版本的原因：
kafka教程-Spring for Apache Kafka版本的选择

application.properties

注意修改2处：

  1. 指定springboot端口号
  2. 指定kafka的ip、port
  3. 自动提交offset
  4. 指定默认消费者组
server.port=8083

###########【Kafka集群】###########
spring.kafka.bootstrap-servers=hadoop1:9092,hadoop2:9092,hadoop3:9092
###########【初始化消费者配置】###########
# 默认的消费组ID
spring.kafka.consumer.properties.group.id=user-group
# 是否自动提交offset
spring.kafka.consumer.enable-auto-commit=true
# 提交offset延时(接收到消息后多久提交offset)
spring.kafka.consumer.auto.commit.interval.ms=1000
# 当kafka中没有初始offset或offset超出范围时将自动重置offset
# earliest:重置为分区中最小的offset;
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
# none:只要有一个分区不存在已提交的offset,就抛出异常;
spring.kafka.consumer.auto-offset-reset=earliest
# 消费会话超时时间(超过这个时间consumer没有发送心跳,就会触发rebalance操作)
spring.kafka.consumer.properties.session.timeout.ms=120000
# 消费请求超时时间
spring.kafka.consumer.properties.request.timeout.ms=180000
# Kafka提供的序列化和反序列化类
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# 消费端监听的topic不存在时，项目启动会报错(关掉)
spring.kafka.listener.missing-topics-fatal=false
# 设置批量消费
# spring.kafka.listener.type=batch
# 批量消费每次最多消费多少条消息
# spring.kafka.consumer.max-poll-records=50

创建消费者类

当有消息时，会自动执行 onMessage() 方法

@Component
public class Consumer {
    // 消费监听
    @KafkaListener(topics = {"userlog"})
    public void onMessage(ConsumerRecord<?, ?> record) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }
}

测试

  1. 启动 zookeeper 服务
    windows下双击 zkServer.cmd 运行

  2. 启动 kafka 服务
    https://www.malaoshi.top/show_1IXLYoFNwKL.html

  3. 启动消费者springboot工程

  4. 启动生产者springboot工程

  5. 访问 http://localhost:8081/kafka/test-msg ，通过Controller向kafka发送消息

  6. 在消费者springboot工程的控制台中，可以收到消息

原文出处：/show_1IXNG2Ao91P.html

吉ICP备15003043号