Springboot 整合kafka

首先搭建两个springboot项目，一个作为生产者，一个作为消费者。kafka可以本地搭建，可以参考https://blog.csdn.net/weixin_42360600/article/details/123048409

生产者

项目结构

1、pom引入kafka依赖

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

2、application.properties中配置kafka生产者

server.port=8080
###########【Kafka集群】###########
spring.kafka.bootstrap-servers=localhost:9092
#==================================【初始化生产者配置】==================================#
# 重试次数
spring.kafka.producer.retries=0
# 应答级别:多少个分区副本备份完成时向生产者发送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
# 自定义分区器
#spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner

3、编写生产者发送消息

@Component
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
    /**
     * 发送消息
     */
    public void sendMessage() {
        try{
            //生产消息
            String message = "hello ！ 测试kafka ";
            ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send("hello","hello", message);
            listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    System.out.println("sendMessage success");
                }
                @Override
                public void onFailure(Throwable ex) {
                    System.out.println("sendMessage error");
                }
            });
        }catch (Exception e){
            System.out.println("sendMessage exception");
        }

    }
}

4、编写controller
页面请求触发kafka生产者发送消息

@RestController
public class KafkaProController {
    @Autowired
    private KafkaProducer kafkaProducer;
    @RequestMapping("/hello")
    public String hello(){
        System.out.println("------->测试生产者发送消息");
        kafkaProducer.sendMessage();
        return "kafka消息已发送.";
    }
}

消费者

项目结构

1、pom引入kafka

<!-- kafka begin -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<!-- kafka end   -->

2、application.properties配置kafka消费者

server.port=8081
###########【Kafka集群】###########
spring.kafka.bootstrap-servers=localhost:9092
#==================================【初始化消费者配置】==================================#
# 默认的消费组ID
spring.kafka.consumer.properties.group.id=defaultConsumerGroup
# 是否自动提交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=latest
# 消费会话超时时间(超过这个时间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

3、编写消费者

@Component
public class TopicComponent {

    @KafkaListener(topics = {"hello"})
    public void handMessage(ConsumerRecord<String, String> record){
        String topic = record.topic();
        String msg = record.value();
        System.out.println("消费者接受消息：topic-->"+topic+",msg->>"+msg);
    }
}

测试

先保证本地zookeeper 、kafka已经启动
启动生产者

启动消费者

页面访问，生产者发送消息

查看后台日志
生产者

消费者

ok，测试成功！结合具体业务编写msg即可