springboot集成kafka的相关配置及自定义

这篇具有很好参考价值的文章主要介绍了springboot集成kafka的相关配置及自定义。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

之前的文章末尾,简单的实现了springboot集成kafka,完成了简单的测试,今天我们来扩展一下相关内容。

首先详解一下配置文件的内容:

spring:
  kafka:
    # 指定 kafka 地址,我这里部署在的虚拟机,开发环境是Windows,kafkahost是虚拟机的地址, 若外网地址,注意修改为外网的IP( 集群部署需用逗号分隔)
    producer:
      bootstrap-servers: 124.223.205.125:9092
      # 发生错误后,消息重发的次数。
      retries: 3
      #当有多个消息需要被发送到同一个分区时,生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小,按照字节数计算。
      batch-size: 16384
      # 设置生产者内存缓冲区的大小。
      buffer-memory: 33554432
      # 指定消息key和消息体的序列化方式
      key-deserializer: org.apache.kafka.common.serialization.StringSerializer
      value-deserializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      properties:
        # 自定义生产者拦截器
        interceptor.classes: com.volga.kafka.interceptor.ProducerPrefixInterceptor
    consumer:
      enable-auto-commit: false #手动提交
      bootstrap-servers: ${spring.kafka.producer.bootstrap-servers}
      # 指定 group_id
      group-id: group_id
      # Kafka中没有初始偏移或如果当前偏移在服务器上不再存在时,默认区最新 ,有三个选项 【latest, earliest, none】
      auto-offset-reset: earliest
      # 指定消息key和消息体的序列化方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        # 自定义消费者拦截器
        interceptor.classes: com.volga.kafka.interceptor.ConsumerPrefixInterceptor
    # 默认主题
    topic: my-topic
    listener:
      # 在侦听器容器中运行的线程数。
      concurrency: 5
      #listner负责ack,每调用一次,就立即commit
      ack-mode: manual_immediate
      missing-topics-fatal: false
      type: batch

以上的producer和consumer的相关配置也可以在java文件中实现:

springboot集成kafka的相关配置及自定义

如上图的实现,自己可以手动实现一下。

kafka内部自己实现分区、策略等一系列的逻辑,当然这些也可以自定义,这里有需要的可以自己研究一下,我这里就不需要了。

接下来,我介绍一个场景,是大家在实际的项目中可以用到。有这样一个场景,在项目中,我们搭建了kafka集群,然而在环境中,会有各种不同的消息,有一些消息不是我们需要的,有些消息是我们需要的,这时我们可以通过过滤器来进行控制和过滤。

在生产者端,我添加一个过滤器在消息前统一加上一个前缀。

 文章来源地址https://www.toymoban.com/news/detail-473852.html

@Slf4j
public class ProducerPrefixInterceptor implements ProducerInterceptor<String,String> {

    AtomicInteger success = new AtomicInteger(0);
    AtomicInteger fail = new AtomicInteger(0);

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        // 消息统一添加前缀
        String modifyValue = "prefix-"+producerRecord.value();
        return new ProducerRecord<>(producerRecord.topic(), producerRecord.partition(), producerRecord.timestamp(), producerRecord.key(), modifyValue, producerRecord.headers());
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if (Objects.nonNull(recordMetadata)){
            success.incrementAndGet();
        }else {
            fail.incrementAndGet();
        }
    }

    @Override
    public void close() {
        log.info("success:%d\nfail:%d\n",success.get(),fail.get());
        success.set(0);
        fail.set(0);
    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}

然后在消费者消费时,我遇到prefix前缀的消息时,就统一过滤掉,这不是我所需要的消息。

@Slf4j
public class ConsumerPrefixInterceptor implements ConsumerInterceptor<String,String> {

    /**
     * 过滤掉test开头的数据
     * @param consumerRecords
     * @return
     */
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> consumerRecords) {
        List<ConsumerRecord<String, String>> filterRecords = new ArrayList<>();
        Map<TopicPartition, List<ConsumerRecord<String, String>>> newRecords= new HashMap<>();
        Set<TopicPartition> partitions = consumerRecords.partitions();
        for(TopicPartition tp : partitions){
            List<ConsumerRecord<String, String>> records = consumerRecords.records(tp);
            for(ConsumerRecord<String, String> record: records){
                if(!record.value().startsWith("prefix")) {
                    filterRecords.add(record);
                }
            }
            if(filterRecords.size() > 0){
                newRecords.put(tp, filterRecords);
            }
        }
        return new ConsumerRecords<>(newRecords);
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> map) {
        map.forEach((k,v) -> log.info("tp:%s--offset:%s\n",k,v));
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}

相关的配置在上述的配置文件中也已经列出了,也可以在代码中加以配置。

props.put("properties.interceptor.classes","com.volga.kafka.interceptor.ProducerPrefixInterceptor");

这样就能实现过滤不需要的消息了。

翻看源码中,生产端主要是KafkaProducer<k,v>这个类中KafkaProducer方法:

springboot集成kafka的相关配置及自定义

消费端这边是对应的:KafkaConsumer<k,v>,这个类中KafkaConsumer方法:

springboot集成kafka的相关配置及自定义

源码里kafka实现的配置逻辑大家可以仔细研究一下。

我是空谷有来人,谢谢支持! 

 

到了这里,关于springboot集成kafka的相关配置及自定义的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • SpringBoot项目集成kafka及常规配置

            使用 spring-kafka 的api,在springboot项目中集成kafka能力,封装配置。 1.1 KafkaConfiguration 公共配置 1.2 KafkaConsumerConfiguration 消费者配置 1.3 KafkaListenerConfiguration 监听配置 1.4 KafkaProducerConfiguration 生产者配置 2.1 ConsumerFactoryBuilder 消费者工厂 2.2 ProducerFactoryBuilder 生产者工

    2024年02月16日
    浏览(39)
  • C++引用的相关定义以及使用场景与指针之前的区别

    🌏博客主页: 主页 🔖系列专栏: C++ ❤️感谢大家点赞👍收藏⭐评论✍️ 😍期待与大家一起进步! 引用不是新定义一个变量,而是给已存在变量取了一个别名,编译器不会为引用变量开辟内存空间,它和它引用的变量共用同一块内存空间。 引用的特性: 引用在定义时必

    2024年02月03日
    浏览(36)
  • springBoot集成caffeine,自定义缓存配置 CacheManager

    目录 springboot集成caffeine Maven依赖 配置信息:properties文件 config配置 使用案例 Caffeine定制化配置多个cachemanager springboot集成redis并且定制化配置cachemanager Caffeine是一种基于 服务器内存的缓存库 。它将数据存储在应用程序的内存中,以实现快速的数据访问和高性能。 由于Caffei

    2024年02月13日
    浏览(32)
  • 【Docker】网络配置及自定义网络的使用

            Docker的网络配置主要是指Docker容器与外部网络之间的连接设置,包括容器内部的IP地址、端口号等。Docker提供了多种网络模式,包括bridge、host、none等,以满足不同的需求。         默认情况下,Docker使用bridge模式,即创建一个虚拟网桥,将容器连接到该网桥上

    2024年01月20日
    浏览(43)
  • SpringBoot【集成 jasypt】实现配置信息自定义加解密(自定义的属性探测和密码解析器)

    Jasypt是一个Java简易加密库,用于加密配置文件中的敏感信息,如数据库密码。它可以帮助开发人员在应用程序中加密密码、敏感信息和数据通信,还包括高安全性、基于标准的加密技术、可同时单向和双向加密的加密密码、文本、数字和二进制文件。如果您正在使用Spring B

    2024年02月05日
    浏览(52)
  • 39、springboot的前端静态资源的WebJar支持(bootstrap、jquery等)及自定义图标和首页

    代码演示: 需求:之前的前端页面比如要使用到一些jquery框架,都是需要把该框架的js资源拷贝到 resources 静态资源路径下面,然后在每个前端页面进行引入。 现在就是通过 Web Jar 的方式,通过在 pom文件今天添加依赖来引入就可以了,不用拷贝js、css那些资源到项目中。 到

    2024年02月11日
    浏览(39)
  • 关于磁盘空间不够,导致报错 springboot内置tomcat相关的临时目录无法创建等问题,如何自定义配置 tomcat 缓存文件路径

    关于磁盘空间不够,导致报错 springboot内置tomcat相关的临时目录无法创建等问题,如何自定义配置 tomcat 缓存文件路径 解决方法: shell 脚本命令:(配置自定义缓存路径参数 -Djava.io.tmpdir ) 注意: tmpPath 的绝对路径值必须是 / 斜杠结尾。 1. springboot内置tomcat相关的临时目录自

    2024年02月09日
    浏览(50)
  • kafka:java集成 kafka(springboot集成、客户端集成)

    摘要 对于java的kafka集成,一般选用springboot集成kafka,但可能由于对接方kafka老旧、kafka不安全等问题导致kafak版本与spring版本不兼容,这个时候就得自己根据kafka客户端api集成了。 一、springboot集成kafka 具体官方文档地址:https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/

    2023年04月22日
    浏览(57)
  • Kafka:springboot集成kafka收发消息

    kafka环境搭建参考Kafka:安装和配置_moreCalm的博客-CSDN博客 1、springboot中引入kafka依赖 2、配置application.yml 传递String类型的消息 3、controller实现消息发送接口 4、component中实现接收类HelloListener  5、测试 浏览器访问该接口并查看控制台         接收成功   传递对象类型的消息

    2024年02月13日
    浏览(41)
  • 八、SpringBoot集成Kafka

    1、修改 SpringBoot 核心配置文件 application.propeties, 添加生产者相关信息 2、创建 controller 从浏览器接收数据, 并写入指定的 topic 1、修改 SpringBoot 核心配置文件 application.propeti 2、创建类消费 Kafka 中指定 topic 的数据 一些配置总结

    2024年02月10日
    浏览(36)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包