标签：Spring Boot private kafka com org import Kafka consumer

前言

Kafka 是一个消息队列产品，基于Topic partitions的设计，能达到非常高的消息发送处理性能。本文主是基于Spirng Boot封装了Apache 的Kafka-client，用于在Spring Boot 项目里快速集成kafka。

一、Kafka 是什么？

Apache Kafka是分布式发布-订阅消息系统。
它最初由LinkedIn公司开发，之后成为Apache项目的一部分。
Kafka是一种快速、可扩展的、设计内在就是分布式的，分区的和可复制的提交日志服务。

二、消息者

1.引入库

引入需要依赖的jar包，引入POM文件

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

2.配置文件

spring:
  custom:
    kafka:
      username: admin
      password: admin-secret
      partitions: 1
      enable-auto-commit: false
      topics: CHANNEL-BodyBusiness-dataDev,CHANNEL-BodyBusiness-pushDev
      groupId: consumer-group-lms-dev
      batch-listener: false
      bootstrap-servers:
        - 192.168.1.95:9092

3.端启动类

启动类名 EnableAutoKafkaClient

package com.cdkjframework.kafka.consumer.annotation;

import com.cdkjframework.kafka.consumer.config.KafkaClientMarkerConfiguration;
import org.springframework.context.annotation.Import;

import java.lang.annotation.*;

/**
 * @ProjectName: cdkj-framework
 * @Package: com.cdkjframework.kafka.consumer.annotation
 * @ClassName: EnableAutoKafkaClient
 * @Description: Kafka客户端自动启动类
 * @Author: xiaLin
 * @Date: 2023/7/18 9:20
 * @Version: 1.0
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import({KafkaClientMarkerConfiguration.class})
public @interface EnableAutoKafkaClient {
}

4.spring.factories配置文件

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
  com.cdkjframework.kafka.consumer.config.KafkaClientAutoConfiguration

5.配置类

5.1 Kafka客户端配置

package com.cdkjframework.kafka.consumer.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

import java.util.List;

/**
 * @ProjectName: cdkj-framework
 * @Package: com.cdkjframework.kafka.consumer.config
 * @ClassName: KafkaClientConfig
 * @Description: Kafka客户端配置
 * @Author: xiaLin
 * @Version: 1.0
 */
@Data
@Configuration
@ConfigurationProperties(prefix = "spring.custom.kafka")
public class KafkaClientConfig {

  /**
   * 服务列表
   */
  private List<String> bootstrapServers;

  /**
   * 主题
   */
  private List<String> topics;

  /**
   * 账号
   */
  private String username;

  /**
   * 密码
   */
  private String password;

  /**
   * 延迟为1毫秒
   */
  private Integer linger = 1;

  /**
   * 批量大小
   */
  private Integer batchSize = 16384;

  /**
   * 重试次数，0为不启用重试机制
   */
  private Integer retries = 0;

  /**
   * 人锁
   */
  private Integer maxBlock = 6000;

  /**
   * acks
   */
  private String acks = "1";

  /**
   * security.providers
   */
  private String securityProviders;

  /**
   * 启用自动提交
   */
  private boolean enableAutoCommit = true;

  /**
   * 会话超时
   */
  private String sessionTimeout = "5000";

  /**
   * 会话超时
   */
  private Integer maxPollInterval = 10000;

  /**
   * 组ID
   */
  private String groupId = "defaultGroup";

  /**
   * 最大投票记录
   */
  private Integer maxPollRecords = 1;

  /**
   * 并发性
   */
  private Integer concurrency = 3;

  /**
   * 拉取超时时间
   */
  private Integer pollTimeout = 60000;

  /**
   * 批量监听
   */
  private boolean batchListener = false;

  /**
   * 副本数量
   */
  private String sort = "1";

  /**
   * 分区数
   */
  private Integer partitions = 3;

  /**
   * 消费者默认支持解压
   */
  private String compressionType = "none";

  /**
   * offset偏移量规则设置
   */
  private String autoOffsetReset = "earliest";

  /**
   * 自动提交的频率
   */
  private Integer autoCommitInterval = 100;

  /**
   * 生产者可以使用的总内存字节来缓冲等待发送到服务器的记录
   */
  private Integer bufferMemory = 33554432;

  /**
   * 消息的最大大小限制
   */
  private Integer maxRequestSize = 1048576;
}

5.2 Kafka客户端自动配置

package com.cdkjframework.kafka.consumer.config;

import com.cdkjframework.kafka.consumer.ConsumerConfiguration;
import com.cdkjframework.kafka.consumer.service.ConsumerService;
import com.cdkjframework.kafka.consumer.listener.ConsumerListener;
import lombok.RequiredArgsConstructor;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.web.reactive.function.client.WebClientAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;

/**
 * @ProjectName: cdkj-framework
 * @Package: com.cdkjframework.kafka.consumer.config
 * @ClassName: KafkaClientAutoConfiguration
 * @Description: Kafka客户端自动配置
 * @Author: xiaLin
 * @Date: 2023/7/18 9:21
 * @Version: 1.0
 */
@Lazy(false)
@RequiredArgsConstructor
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties(KafkaClientConfig.class)
@AutoConfigureAfter({WebClientAutoConfiguration.class})
@ImportAutoConfiguration(ConsumerConfiguration.class)
@ConditionalOnBean(KafkaClientMarkerConfiguration.Marker.class)
public class KafkaClientAutoConfiguration {

  /**
   * 消费者服务接口
   */
  private final ConsumerService consumerService;

  /**
   * kafka topic 启动触发器
   *
   * @return 返回结果
   */
  @Bean
  @ConditionalOnMissingBean
  public ConsumerListener kafkaConsumer() {
    return new ConsumerListener(consumerService);
  }
}

5.3 Kafka客户端标记配置

package com.cdkjframework.kafka.consumer.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;

/**
 * @ProjectName: cdkj-framework
 * @Package: com.cdkjframework.kafka.consumer.config
 * @ClassName: KafkaClientMarkerConfiguration
 * @Description: Kafka客户端标记配置
 * @Author: xiaLin
 * @Date: 2023/12/6 13:11
 * @Version: 1.0
 */
@EnableKafka
@Configuration(proxyBeanMethods = false)
public class KafkaClientMarkerConfiguration {

  @Bean
  public Marker kafkaMarker() {
    return new Marker();
  }

  public static class Marker {

  }
}

6.消费者配置

package com.cdkjframework.kafka.consumer;

import com.cdkjframework.kafka.consumer.config.KafkaClientConfig;
import com.cdkjframework.util.log.LogUtils;
import com.cdkjframework.util.tool.JsonUtils;
import com.cdkjframework.util.tool.StringUtils;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.Consumer;
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.common.TopicPartition;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.BatchErrorHandler;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ConsumerAwareErrorHandler;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 * @ProjectName: cdkj-framework
 * @Package: com.cdkjframework.kafka.consumer
 * @ClassName: ProducerConfiguration
 * @Description: 设置@Configuration、@EnableKafka两个注解，声明Config并且打开KafkaTemplate能力。
 * @Author: xiaLin
 * @Version: 1.0
 */
@Configuration
@RequiredArgsConstructor
public class ConsumerConfiguration {

  /**
   * 日志
   */
  private final LogUtils logUtils = LogUtils.getLogger(ConsumerConfiguration.class);

  /**
   * JAAS配置
   */
  private String JAAS_CONFIG = "org.apache.kafka.common.security.plain.PlainLoginModule required username=%s password=%s;";

  /**
   * 配置
   */
  private final KafkaClientConfig kafkaClientConfig;

  /**
   * 监听容器工厂
   *
   * @return 返回结果
   */
  @Bean(name = "kafkaListenerContainerFactory")
  public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String>
            factory = new ConcurrentKafkaListenerContainerFactory<>();
    // 设置消费者工厂
    factory.setConsumerFactory(consumerFactory());
    // 消费者组中线程数量
    factory.setConcurrency(kafkaClientConfig.getConcurrency());
    // 拉取超时时间
    factory.getContainerProperties().setPollTimeout(kafkaClientConfig.getPollTimeout());
    // 当使用批量监听器时需要设置为true
    factory.setBatchListener(kafkaClientConfig.isBatchListener());
    // 将单条消息异常处理器添加到参数中
    factory.setErrorHandler(new ConsumerAwareErrorHandler() {
      @Override
      public void handle(Exception thrownException, ConsumerRecord<?, ?> data, Consumer<?, ?> consumer) {
        logUtils.error("// 将单条消息异常：" + thrownException.getMessage());
        logUtils.error("// 将单条消息：" + data.toString() + "，" + consumer.toString());
        Iterator<TopicPartition> iterator = consumer.assignment().iterator();
        if (iterator.hasNext()) {
          // 提交重新消费
          consumer.seek(iterator.next(), data.offset());
        }
      }
    });
    if (kafkaClientConfig.isBatchListener()) {
      // 将批量消息异常处理器添加到参数中
      factory.setBatchErrorHandler(new BatchErrorHandler() {
        @Override
        public void handle(Exception thrownException, ConsumerRecords<?, ?> data) {
          logUtils.error("// 将批量消息异常：" + thrownException.getMessage());
          logUtils.error(thrownException);
          logUtils.error(JsonUtils.objectToJsonString(data));
        }
      });
    }
//        factory.setContainerCustomizer();

    return factory;
  }

  /**
   * 消费者工厂
   *
   * @return 返回消费工厂
   */
  @Bean
  public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfig());
  }

  /**
   * 消费者配置
   *
   * @return 返回结果
   */
  @Bean
  public Map<String, Object> consumerConfig() {
    Map<String, Object> propsMap = new HashMap<>();
    // Kafka地址
    propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaClientConfig.getBootstrapServers());
    //配置默认分组，这里没有配置+在监听的地方没有设置groupId，多个服务会出现收到相同消息情况
    propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaClientConfig.getGroupId());
    // 是否自动提交offset偏移量(默认true)
    propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaClientConfig.isEnableAutoCommit());
    // 心跳机制
    propsMap.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaClientConfig.getMaxPollInterval());
    // 每次读取最大记录
    propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaClientConfig.getMaxPollRecords());
    if (kafkaClientConfig.isEnableAutoCommit()) {
      // 自动提交的频率(ms)
      propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaClientConfig.getAutoCommitInterval());
    }
    // 键的反序列化方式
    propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    // 值的反序列化方式
    propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    // offset偏移量规则设置：
    // (1)、earliest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
    // (2)、latest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
    // (3)、none：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
    propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaClientConfig.getAutoOffsetReset());

    // 安全认证 账号密码
    if (StringUtils.isNotNullAndEmpty(kafkaClientConfig.getUsername()) &&
            StringUtils.isNotNullAndEmpty(kafkaClientConfig.getPassword())) {
      propsMap.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_PLAINTEXT.name);
      String SASL_MECHANISM = "PLAIN";
      propsMap.put(SaslConfigs.SASL_MECHANISM, SASL_MECHANISM);
      propsMap.put(SaslConfigs.SASL_JAAS_CONFIG, String.format(JAAS_CONFIG, kafkaClientConfig.getUsername(), kafkaClientConfig.getPassword()));
    }

    return propsMap;
  }
}

7.消费者服务

package com.cdkjframework.kafka.consumer.service;

import com.cdkjframework.exceptions.GlobalException;

/**
 * @ProjectName: cdkj-framework
 * @Package: com.cdkjframework.kafka.consumer
 * @ClassName: com.cdkjframework.kafka.consumer.service.ConsumerService
 * @Description: 消费者服务
 * @Author: xiaLin
 * @Version: 1.0
 */
public interface ConsumerService {

    /**
     * 消息内容
     *
     * @param topics  主题
     * @param message 内容
     * @throws GlobalException 异常信息
     */
    void onMessage(String topics, String message) throws GlobalException;
}

8.消费者监听器

package com.cdkjframework.kafka.consumer.listener;

import com.cdkjframework.kafka.consumer.service.ConsumerService;
import com.cdkjframework.util.log.LogUtils;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;

/**
 * @ProjectName: cdkj-kafka-client
 * @Package: com.cdkjframework.kafka.consumer
 * @ClassName: com.cdkjframework.kafka.consumer.listener.ConsumerListener
 * @Description: 消费者监听器
 * @Author: xiaLin
 * @Version: 1.0
 */
public class ConsumerListener {

    /**
     * 日志
     */
    private final LogUtils logUtils = LogUtils.getLogger(ConsumerListener.class);

    /**
     * 消费者服务接口
     */
    private final ConsumerService consumerService;

    /**
     * 构造函数
     *
     * @param consumerService 消费者服务接口
     */
    public ConsumerListener(ConsumerService consumerService) {
        this.consumerService = consumerService;
    }

    /**
     * 单条监听MQ消息
     *
     * @param data     消息内容
     * @param consumer 消息者
     */
    @KafkaListener(topics = "#{'${spring.custom.kafka.topics}'.split(',')}", groupId = "${spring.custom.kafka.groupId}")
    public void listener(ConsumerRecord<String, String> data, Consumer consumer) {
        try {
            consumerService.onMessage(data.topic(), data.value());
            consumer.commitAsync();
        } catch (Exception e) {
            logUtils.error(e);
            // 抛出异常，以重试消费
            throw new RuntimeException(e.getMessage());
        }
    }
}

总结

例如：以上就是今天要讲的内容，本文仅仅简单介绍了 Spring Boot 集成消息消费者的封装。

相对应的开源项目欢迎访问：维基框架

标签：Spring,Boot,private,kafka,com,org,import,Kafka,consumer
From： https://www.cnblogs.com/cdkj/p/17982232

第三章 Spring Boot 整合 Kafka消息队列消息者

前言

一、Kafka 是什么？

二、消息者

1.引入库

2.配置文件

3.端启动类

4.spring.factories配置文件

5.配置类

5.1 Kafka客户端配置

5.2 Kafka客户端自动配置

5.3 Kafka客户端标记配置

6.消费者配置

7.消费者服务

8.消费者监听器

总结

相关文章

赞助商

阅读排行

第三章 Spring Boot 整合 Kafka消息队列 消息者

前言

一、Kafka 是什么？

二、消息者

1.引入库

2.配置文件

3.端启动类

4.spring.factories配置文件

5.配置类

5.1 Kafka客户端配置

5.2 Kafka客户端自动配置

5.3 Kafka客户端标记配置

6.消费者配置

7.消费者服务

8.消费者监听器

总结

相关文章

赞助商

阅读排行

第三章 Spring Boot 整合 Kafka消息队列消息者