kafka:spring集成 kafka(springboot集成客户端集成)

Posted 崛起之路

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了kafka:spring集成 kafka(springboot集成客户端集成)相关的知识,希望对你有一定的参考价值。

摘要

对于java的kafka集成,一般选用springboot集成kafka,但可能由于对接方kafka老旧、kafka不安全等问题导致kafak版本与spring版本不兼容,这个时候就得自己根据kafka客户端api集成了。

一、springboot集成kafka

具体官方文档地址:https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/

1、加入依赖,spring-boot-starter-web和spring-kafka 的版本号可以看它们依赖的spring版本是否一致,这里pom依赖如下:

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>2.7.9</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.9.6</version>
        </dependency>

2、添加application.yml配置,具体如下:

server:
  port: 8087
spring:
  mvc:
    pathmatch:
      matching-strategy: ant_path_matcher
  kafka:
    bootstrap-servers: 192.168.189.128:9092,92.168.189.128:9093,192.168.189.128:9094
    consumer:
      properties:
        group:
          id: boot-kafka

3、发送消息,由于KafkaTemplate是自动装配的,所以只要在spring的bean里注入KafkaTemplate发送消息即可,具体如下:

package com.longqi.bootkafka.controller;

import com.longqi.bootkafka.entity.MessageParam;
import com.longqi.bootkafka.entity.Wrapper;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.validation.Valid;

/**
 * <p>
 * 测试 前端控制器
 * </p>
 * @author LongQi
 * @since 2021-06-23
 */

@Slf4j
@RestController
@RequestMapping("/test")
@Api(value = "TestController", tags = "测试 API")
public class TestController 

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private Boolean isSend = true;

    @PostMapping("/kafka/sendMessage")
    @ApiOperation(httpMethod = "POST", value = "发送kafka告警消息", response = Wrapper.class)
    public Wrapper sendKafkaMessage(@Valid @ApiParam("参数") @RequestBody MessageParam param) 
        kafkaTemplate.send(param.getTopic(), param.getMessage());
        return Wrapper.ok(true);
    

这里用参数"message": "asd54a6d46a4ds","topic": "device-alarm-test"进行测试,会报如下日志:

发现会报警告:[Producer clientId=producer-1] Error while fetching metadata with correlation id 34 : device-alarm-test=LEADER_NOT_AVAILABLE,获取主题元数据错误,这个可以忽略,查找元数据失败,spring-kafka会自动创建主题的,后续再次发送消息,是不会报这个错误的。

查看可视化工具EFAK,发现主题device-alarm-test是自动创建成功,分区数是kafka的集群配置service.properties里配置的分区9,具体如下:

可以看到,其中一个分区保存了这个消息,logsize变成了1,说明这个消息是发送成功的。另外也可以看到主题的各分区主备消息所在的节点是不一样的。

4、接收消息,接收消息也很简单,只要在spring的bean里使用KafkaListener注解即可,具体如下:

可视化工具也能看到该主题该消费者9个分区的消费情况,具体如下:

logSize为存入分区parttion消息数量,Offset为消费的偏移量(已消费的数量),Lag为未消费的数量(积压的数量),Owner为消费者,目前可以看到消费者为同一个,即只有1个线程在消费这9个分区的消息。

二、客户端集成kafka

直接使用kafka客户端,建议使用最新版的客户端,毕竟没有其他框架版本限制,能用最新的就用最新的,毕竟新的一般性能强也修复了bug。好比23年2月份出现的kafka安全漏洞:远程代码执行漏洞CVE-2023-25194,对现在最新版3.4.0无效,对以前大部分版本就有效。

1、添加依赖,具体如下:

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.4.0</version>
        </dependency>

2、发送和消费消息,具体代码如下:

package com.longqi.bootkafka.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author LongQi
 * @projectName boot-integration
 * @description: kafka配置
 * @date 2023/3/13 14:42
 */

public class KafkaConfig 

    public static void main(String[] args) 
        // 声明主题
        String topic = "device-alarm-test";
        // 创建消费者
        Properties consumerConfig = new Properties();
        consumerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.189.128:9092,92.168.189.128:9093,192.168.189.128:9094");
        consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG,"boot-kafka");
        consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer kafkaConsumer = new KafkaConsumer(consumerConfig);
        // 订阅主题并循环拉取消息
        kafkaConsumer.subscribe(Arrays.asList(topic));
        new Thread(new Runnable() 
            @Override
            public void run() 
                while (true)
                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(10000));
                    for(ConsumerRecord<String, String> record:records)
                        System.out.println(record.value());
                    
                
            
        ).start();
        // 创建生产者
        Properties producerConfig = new Properties();
        producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.189.128:9092,92.168.189.128:9093,192.168.189.128:9094");
        producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG,"boot-kafka-client");
        producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer producer = new KafkaProducer<>(producerConfig);
        // 给主题发送消息
        producer.send(new ProducerRecord<>(topic, "hello,"+System.currentTimeMillis()));
    

最后可以看到打印消息如下:

成功接收到消息并打印

以上是关于kafka:spring集成 kafka(springboot集成客户端集成)的主要内容,如果未能解决你的问题,请参考以下文章

Spring 集成Kafka(完整版)

spring集成kafka

Spring boot 集成Kafka

spring集成kafka

spring集成kafka运行时报错:Failed to construct kafka producer] with root cause

spring boot集成kafka开发,接收kafka消息,Java