淘先锋技术网

首页 1 2 3 4 5 6 7

Apache Kafka - ConsumerInterceptor 实战 (1)

文章目录


在这里插入图片描述

概述

ConsumerInterceptor是Kafka中的一个重要组件，它允许开发人员在Kafka消费者端拦截和修改消息的处理过程。ConsumerInterceptor可以用于实现各种功能，从消息监控到数据转换和错误处理，为开发人员提供了更大的灵活性和可定制性。

ConsumerInterceptor的主要作用是在消息被消费之前和之后对其进行拦截和处理。它可以用于以下几个方面：

  1. 监控：通过ConsumerInterceptor，可以在消息被消费之前和之后记录和监控消息的元数据，例如消息的偏移量、主题、分区等信息。这对于跟踪和分析消息流的健康状况以及性能优化非常有用。

  2. 转换：ConsumerInterceptor还可以用于对消息进行转换和修改。通过拦截消息并对其进行操作，可以在消费者端对消息进行格式转换、数据解析或者其他自定义处理。例如，你可以将消息从一种格式转换为另一种格式，或者对消息进行特定的业务处理。

  3. 错误处理：当消费者在处理消息时发生错误或异常情况时，ConsumerInterceptor可以捕获这些错误并采取适当的措施。你可以在拦截器中实现自定义的错误处理逻辑，例如记录错误日志、发送告警通知或者进行重试操作，从而提高应用程序的可靠性和容错性。

使用场景

使用场景方面，ConsumerInterceptor可以在多种情况下发挥作用，例如：

  1. 监控和统计：你可以使用ConsumerInterceptor来收集和记录消费者端的统计信息，例如消费速率、处理延迟等。这样可以帮助你监控应用程序的性能并进行性能优化。

  2. 数据转换：如果你需要将消息从一种格式转换为另一种格式，例如将JSON消息转换为Avro格式，你可以使用ConsumerInterceptor来实现这个转换过程。

  3. 数据验证：ConsumerInterceptor可以用于验证消息的有效性和完整性。你可以在拦截器中实现验证逻辑，例如检查消息的签名或者校验消息的结构，以确保只有符合要求的消息被消费。

  4. 错误处理和重试：当消费者在处理消息时遇到错误，例如数据库连接失败或者网络故障，你可以使用ConsumerInterceptor来捕获这些错误并采取适当的措施。你可以在拦截器中实现自定义的错误处理逻辑，例如记录错误日志、发送告警通知或者进行消息重试。

总之，ConsumerInterceptor为开发人员提供了在消费者端对消息进行拦截、处理和定制的能力。通过使用ConsumerInterceptor，你可以实现一系列功能，包括监控、数据转换和错误处理，从而更好地控制和管理Kafka消费者端的消息处理过程。

实战

在这里插入图片描述

配置文件

spring:
  kafka:
    bootstrap-servers: 20.10.110.137:9888 # Kafka服务的地址
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer # key的序列化器
      value-serializer: org.apache.kafka.common.serialization.StringSerializer # value的序列化器
      acks: 1 # acks=0：表示producer不需要等待任何确认收到的信息。副本将立即加到socket缓冲区并认为已经发送。如果使用此选项，则存在丢失数据的风险，因为服务器在数据到达副本之前可能会崩溃。
      retries: 0 # 失败重试次数，0表示不启用重试机制
      batch-size: 16384 # 发送缓冲区大小，按照字节计算
      linger-ms: 1 # 发送延时，单位毫秒
      buffer-memory: 33554432 # 发送缓存区的大小，按照字节计算
      compression-type: gzip # 压缩类型，默认是none，可选snappy、gzip、lz4
    consumer:
      #Kafka中没有初始偏移或如果当前偏移在服务器上不再存在时,默认区最新 ，有三个选项 【latest, earliest, none】
      auto-offset-reset: earliest
      #是否开启自动提交
      enable-auto-commit: false
      #key的解码方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      #value的解码方式
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      #消费者组groupid
      group-id: process-group
      #消费者最大拉取的消息数量
      max-poll-records: 2000
      #消费者最大等待时间
      max-poll-interval-ms: 2000
    listener:
      type: batch
      ack-mode: manual # 手动提交
      concurrency: 12 # 并发数

配置类

package net.zf.module.system.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
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.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ContainerProperties;

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

/**
 * @author artisan
 */
@Slf4j
@Configuration
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServer;

    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;

    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private String enableAutoCommit;

    @Value("${spring.kafka.consumer.key-deserializer}")
    private String keyDeserializer;

    @Value("${spring.kafka.consumer.value-deserializer}")
    private String valueDeserializer;

    @Value("${spring.kafka.consumer.group-id}")
    private String group_id;

    @Value("${spring.kafka.consumer.max-poll-records}")
    private String maxPollRecords;

    @Value("${spring.kafka.consumer.max-poll-interval-ms}")
    private String maxPollIntervalMs;

    @Value("${spring.kafka.listener.concurrency}")
    private Integer concurrency;

    private final String consumerInterceptor = "net.zf.module.system.kafka.interceptor.FailureRateInterceptor";


    /**
     * 消费者配置信息
     */
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>(32);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServer);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,enableAutoCommit);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,keyDeserializer);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,valueDeserializer);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,maxPollRecords);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,maxPollIntervalMs);
        props.put(ConsumerConfig.GROUP_ID_CONFIG,group_id);
        props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,consumerInterceptor );
        return props;
    }


    /**
     * 消费者批量工厂
     */
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> batchFactory() {
        ConcurrentKafkaListenerContainerFactory<String,String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.setBatchListener(true);
        factory.setConcurrency(concurrency);
        return factory;
    }




    /**
     * 异常处理器
     *
     * @return
     */
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareListenerErrorHandler() {
        return (message, exception, consumer) -> {
//            log.error("消息{} , 异常原因{}", message, exception.getMessage());
            log.error("consumerAwareListenerErrorHandler called");

            return null;
        };
    }

}

这段代码是一个用于配置Kafka消费者的Spring配置类。它使用了Spring Kafka库来设置Kafka的消费者配置和相关的监听器。

以下是代码的主要部分的解释：

  1. 通过@Configuration注解将该类标记为一个Spring配置类。
  2. 使用@Value注解注入配置属性，这些属性来自于应用的配置文件（比如application.properties）。
  3. consumerConfigs()方法创建了一个包含Kafka消费者配置信息的props对象，并将其返回。这些配置包括Kafka服务器地址、消费者组ID、序列化/反序列化类等。
  4. batchFactory()方法创建了一个ConcurrentKafkaListenerContainerFactory对象，并设置了相关的属性。它使用了前面定义的消费者配置，并设置了批量消费和并发处理的参数。
  5. consumerAwareListenerErrorHandler()方法创建了一个ConsumerAwareListenerErrorHandler对象，用于处理消费过程中出现的异常。在这个例子中，它只是打印了错误日志。

总体而言，这段代码的目的是配置Kafka消费者的相关属性，包括连接到Kafka服务器的配置、消费者组ID、序列化/反序列化类等。它还定义了一个批量消费的监听器工厂和一个异常处理器。这些配置可以通过注入KafkaListenerContainerFactory和ConsumerAwareListenerErrorHandler来在应用中使用。

自定义ConSumerInterceptor

package net.zf.module.system.kafka.interceptor;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.stereotype.Component;

import java.util.Map;


/**
 * @author artisan
 */

@Slf4j
@Component
public class FailureRateInterceptor implements ConsumerInterceptor<Object, Object> {

    /**
     * 消息消费前的拦截处理
     *
     * @param consumerRecords
     * @return
     */
    @Override
    public ConsumerRecords<Object, Object> onConsume(ConsumerRecords<Object, Object> consumerRecords) {
        // TODO
        log.info("FailureRateInterceptor#onConsume");

        // 根据设定的规则计算失败率，并进行判断是否跳过消息的消费
        // 返回ConsumerRecords对象, 继续执行下游的消费逻辑或者直接返回空的ConsumerRecords对象 (ConsumerRecords.EMPTY)

        return consumerRecords;
    }

    /**
     * 消息提交前进行拦截处理
     *
     * @param map
     */
    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> map) {
        log.info("FailureRateInterceptor#onCommit");
    }


    /**
     * 拦截器关闭前进行拦截处理(如果有的话)
     */
    @Override
    public void close() {
        log.info("FailureRateInterceptor#close");
    }

    /**
     * 初始化配置(如果有的话)
     *
     * @param map
     */
    @Override
    public void configure(Map<String, ?> map) {
        log.info("FailureRateInterceptor#configure");
    }
}

onConsume 可以控制 ConsumerRecords， 通过返回null ，可以暂停消费。

这段代码是一个自定义的Kafka消费者拦截器，实现了ConsumerInterceptor接口。拦截器可以在消息消费和提交的过程中插入自定义的逻辑，用于处理消息或拦截操作。

以下是代码的主要部分的解释：

  1. @Slf4j注解用于自动生成日志记录器。
  2. @Component注解将该类标记为Spring组件，使得它可以被自动扫描和注入到应用中。
  3. 实现了ConsumerInterceptor接口，并重写了其中的方法。
    • onConsume()方法在消费者消费消息之前被调用。在这个例子中，它只是打印了日志信息，表示拦截器的执行。
    • onCommit()方法在消息提交之前被调用。在这个例子中，它只是打印了日志信息，表示拦截器的执行。
    • close()方法在拦截器关闭之前被调用。在这个例子中，它只是打印了日志信息，表示拦截器的执行。
    • configure()方法在拦截器初始化配置时被调用。在这个例子中，它只是打印了日志信息，表示拦截器的执行。
  4. 拦截器的具体逻辑还没有实现，而是用// TODO标记了需要填充的部分。根据注释的描述，它可能会根据设定的规则计算消费失败率，并根据判断跳过或继续消费消息。

总体而言，这段代码定义了一个自定义的Kafka消费者拦截器。拦截器可以在消息消费和提交的过程中执行自定义的逻辑。在这个例子中，拦截器的逻辑还没有实现，只是打印了日志信息以表示拦截器的执行。你需要根据需求实现onConsume()方法中的拦截逻辑，以便根据设定的规则处理消息消费的失败率。

使用

package net.zf.module.system.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import net.zf.module.system.entity.AttackMessage;
import net.zf.module.system.executors.factory.MessageExecutorFactory;
import net.zf.module.system.service.es.AttackMessageESService;
import net.zf.module.system.util.constants.KafkaTopicConstant;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * @author artisan
 */
@Component
@Slf4j
public class AttackKafkaConsumer {

    @Autowired
    private MessageExecutorFactory messageExecutorFactory;

    @Autowired
    private AttackMessageESService attackMessageESService;


    @KafkaListener(topicPattern = KafkaTopicConstant.ATTACK_MESSAGE + ".*",
            containerFactory = "batchFactory",
            errorHandler = "consumerAwareListenerErrorHandler")
    public void processMessage(List<String> records, Acknowledgment ack)  {
        log.info("AttackKafkaConsumer 当前线程 {} , 本次拉取的数据总量:{} ", Thread.currentThread().getId(), records.size());
        try {
            List<AttackMessage> attackMessages = new ArrayList();
            records.stream().forEach(record -> {
                messageExecutorFactory.process(KafkaTopicConstant.ATTACK_MESSAGE).execute(record, attackMessages);
            });
            if (!attackMessages.isEmpty()) {
                String response = attackMessageESService.addDocuments(attackMessages, false);
                log.info("AttackKafkaConsumer本次处理的数据总量:{}, 响应结果: {}", attackMessages.size(), response);
            }
        } finally {
            ack.acknowledge();
        }
    }
}

这段代码定义了一个名为AttackKafkaConsumer的类，它是一个Kafka消费者。它使用了Spring Kafka提供的@KafkaListener注解来指定消费者的相关配置。

以下是代码的主要部分的解释：

  1. @Component注解将该类标记为Spring组件，使得它可以被自动扫描和注入到应用中。
  2. @Slf4j注解用于自动生成日志记录器。
  3. AttackKafkaConsumer类中注入了MessageExecutorFactory和AttackMessageESService两个依赖，通过@Autowired注解实现自动注入。
  4. @KafkaListener注解标记了processMessage()方法作为Kafka消费者的消息处理方法。
    • topicPattern属性指定了要监听的Kafka主题的模式，使用了常量KafkaTopicConstant.ATTACK_MESSAGE并结合通配符.*。
    • containerFactory属性指定了用于创建Kafka监听容器的工厂bean的名称，使用了名为batchFactory的工厂。
    • errorHandler属性指定了用于处理消费者异常的错误处理器的bean的名称，使用了名为consumerAwareListenerErrorHandler的错误处理器。
  5. processMessage()方法是消息的实际处理逻辑。它接收一个List<String>类型的消息记录和一个Acknowledgment对象作为参数。
    • 首先，它记录了当前线程ID和本次拉取的数据总量的日志信息。
    • 然后，它创建了一个空的AttackMessage列表，用于存储处理后的消息。
    • 使用records.stream().forEach()遍历每条消息记录，并通过messageExecutorFactory调用process()方法来处理每条记录，同时将处理结果添加到attackMessages列表中。
    • 在处理完所有消息后，如果attackMessages列表不为空，将调用attackMessageESService的addDocuments()方法将消息添加到Elasticsearch中，并记录处理的数据总量和响应结果的日志信息。
    • 最后，在finally块中调用ack.acknowledge()手动确认消费完成。

总体而言，这段代码定义了一个Kafka消费者类AttackKafkaConsumer，并使用@KafkaListener注解指定了监听的主题、容器工厂和错误处理器。processMessage()方法是处理消息的具体逻辑，它遍历消息记录并调用适当的执行器进行处理，最后将处理结果添加到列表中，并通过Elasticsearch服务将消息存储到数据库中。消费完成后，手动确认消息的消费。

在这里插入图片描述