探索 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
    }
}
</string></string></string></string>

創建消費者應用程式

建立一個簡單的消費者應用程序，用於偵聽 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());
    }
}
</string>

綁定介面

定義一個接口，用於將輸入通道綁定到 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中文網其他相關文章！