kafka教程-创建springboot工程，实现简单消费者（大数据）

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

说明

创建springboot工程

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

添加依赖

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

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

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

application.properties

在 resources 目录下创建 application.properties 文件

注意修改以下几处：

  1. 指定springboot端口号
  2. 指定kafka的ip、port
  3. 指定默认消费者组
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。指定一个新用户组，第一次启动会拉取所有的数据，记住offset。再次启动是从offset位置拉取数据
# latest:重置为分区中最新的offset(消费分区中新产生的数据)。指定一个新用户组，第一次启动，指向最新的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() 方法

package com.std.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;


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

另一种写法

注意：不要同时定义多个接收方法

@KafkaListener(topics = {"userlog"})
public void onMessage(@Payload String message,
                      @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                      @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                      @Header(KafkaHeaders.OFFSET) String offset) {

    // 消费的哪个topic、partition的消息,打印出消息内容
    System.out.println(String.format("消费：topic:%s-partition:%s-offset:%s-value:%s", topic,partition,offset,message));
}

启动类

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Main {
    public static void main(String[] args) {
        SpringApplication.run(Main.class,args);
    }
}

kafka准备

启动 zookeeper 服务

见 链接

启动 kafka 服务

见 链接

创建主题

bin/kafka-topics.sh --bootstrap-server hadoop1:9092 --create --topic userlog

启动消费者

cd /program/kafka_2.12-2.4.1

bin/kafka-console-consumer.sh --bootstrap-server hadoop1:9092 --topic userlog

测试

  1. 启动消费者springboot工程

  2. 启动生产者springboot工程

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

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

原文出处：/show_1IX3273Ncuu3.html

吉ICP备15003043号