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

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

说明

创建springboot工程

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

添加依赖

springboot有对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 文件，内容如下：

注意修改2处：

  1. 指定springboot端口号
  2. 指定kafka的ip、port

server.port=8081

###########【Kafka集群】###########
spring.kafka.bootstrap-servers=hadoop1:9092,hadoop2:9092,hadoop3:9092
###########【初始化生产者配置】###########
# 重试次数
spring.kafka.producer.retries=3
# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
spring.kafka.producer.acks=1
# 批量大小
spring.kafka.producer.batch-size=16384
# 提交延时
spring.kafka.producer.properties.linger.ms=0
# 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
# linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了

# 生产端缓冲区大小
spring.kafka.producer.buffer-memory = 33554432
# Kafka提供的序列化和反序列化类
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

启动时创建topic（不需要）

创建配置类，启动时创建topic

@Configuration
public class KafkaConfig {

    // 创建一个名为testtopic的Topic并设置分区数为8，分区副本数为2
    @Bean
    public NewTopic initialTopic() {
        return new NewTopic("userlog", 8, (short) 2);
    }

    // 如果要修改分区数，只需修改配置值重启项目即可
    // 修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小
    @Bean
    public NewTopic updateTopic() {
        return new NewTopic("userlog", 10, (short) 2);

    }
}

创建Controller

通过该Controller，可以向kafka的 userlog 主题发送消息

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;


@RestController
public class ProducerCtrl {
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    // 发送消息
    @GetMapping("/kafka/{message}")
    public String sendMessage1(@PathVariable("message") String msg) {
        kafkaTemplate.send("userlog", msg);//向 userlog 主题发送消息
        return "消息已发送";
    }
}

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. 访问 http://localhost:8081/kafka/test-msg ，通过Controller向kafka发送消息

  3. 在消费者控制台中，可以看到消息

原文出处：/show_1IX31r3MfMPp.html

吉ICP备15003043号