探索 Spring Cloud Stream Kafka Binder 消费者拦截器

介绍

Spring Cloud Stream 是一个框架，通过抽象 Apache Kafka 和 RabbitMQ 等消息代理来简化消息驱动的微服务的开发。 Spring Cloud Stream 的强大功能之一是它能够与 Kafka 无缝集成，使开发人员能够构建健壮且可扩展的事件驱动应用程序。 Spring Cloud Stream 中的 Kafka Binder 提供了一种轻松连接 Kafka 主题的方法。

在本博客中，我们将深入研究如何将消费者拦截器与 Spring Cloud Stream Kafka Binder 结合使用。 Kafka 中的拦截器提供了一种在应用程序使用记录之前拦截和更改记录的机制，为日志记录、指标收集和数据操作提供了机会。

先决条件

在深入了解详细信息之前，请确保您满足以下先决条件：

• Java 开发工具包 (JDK) 8 或更高版本
• 阿帕奇卡夫卡
• Spring Boot 2.x 或更高版本
• Maven 或 Gradle

设置 Spring Boot 应用程序

首先，让我们设置一个简单的 Spring Boot 项目，其中包含 Spring Cloud Stream 和 Kafka 的必要依赖项。

Maven pom.xml

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>Hoxton.SR10</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

Gradle 构建.gradle

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter'
    implementation 'org.springframework.cloud:spring-cloud-starter-stream-kafka'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
}

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:Hoxton.SR10"
    }
}

配置Kafka Binder

接下来，在 application.yml 文件中配置 Kafka Binder。

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: my-topic
          group: my-group
          consumer:
            interceptor-classes: com.example.MyConsumerInterceptor
      kafka:
        binder:
          brokers: localhost:9092

创建 Kafka 消费者拦截器

要创建消费者拦截器，请实现Kafka提供的ConsumerInterceptor接口。此接口允许您定义自定义逻辑，以便在记录到达应用程序之前拦截和处理记录。

package com.example;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.Configurable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class MyConsumerInterceptor implements ConsumerInterceptor<String, String>, Configurable {

    private static final Logger logger = LoggerFactory.getLogger(MyConsumerInterceptor.class);

    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        records.forEach(record -> {
            logger.info("Intercepted record: key = {}, value = {}", record.key(), record.value());
            // Add your custom logic here
        });
        return records;
    }

    @Override
    public void onCommit(Map offsets) {
        // Custom logic on commit
    }

    @Override
    public void close() {
        // Cleanup resources if necessary
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // Configuration logic
    }
}

创建消费者应用程序

创建一个简单的消费者应用程序，用于侦听来自 Kafka 主题的消息。

package com.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;

@SpringBootApplication
@EnableBinding(KafkaProcessor.class)
public class KafkaConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaConsumerApplication.class, args);
    }

    @StreamListener("input")
    public void handle(Message<String> message) {
        System.out.println("Received message: " + message.getPayload());
    }
}

绑定接口

定义一个接口，用于将输入通道绑定到 Kafka 主题。

package com.example;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;

public interface KafkaProcessor {
    String INPUT = "input";

    @Input(INPUT)
    SubscribableChannel input();
}

运行应用程序

1. 启动 Kafka Broker 并创建所需的主题（my-topic）。
2. 运行 Spring Boot 应用程序。

当向 Kafka 主题生成消息时，MyConsumerInterceptor 将拦截记录，您应该看到拦截的日志消息。

结论

在本博客中，我们探索了如何将消费者拦截器与 Spring Cloud Stream Kafka Binder 结合使用。拦截器提供了一种在应用程序使用记录之前对其进行处理、记录和操作的强大方法。通过集成自定义拦截器，您可以增强 Kafka 消费者的功能，添加日志记录、指标收集和数据转换等有价值的功能。

通过遵循本指南中概述的步骤，您应该能够在 Spring Cloud Stream 应用程序中无缝地实现和配置消费者拦截器。快乐编码！

以上是探索 Spring Cloud Stream Kafka Binder 消费者拦截器的详细内容。更多信息请关注PHP中文网其他相关文章！