如何配置Spring应用的StreamBridge不依赖Schema Registry？

要让StreamBridge不依赖Schema Registry工作，核心是绕过Spring Cloud Stream默认的Schema相关序列化逻辑，改用Kafka原生序列化器，并清理不必要的Schema Registry依赖。以下是具体实现步骤：

1. 清理Schema Registry相关依赖 #

检查项目依赖管理文件（Maven的pom.xml或Gradle的build.gradle），移除或排除所有与Schema Registry相关的依赖，例如：

• spring-cloud-stream-schema-registry-client
• spring-cloud-schema-registry-server

如果是通过Spring Cloud Stream starter间接引入相关依赖，以Maven为例添加排除规则：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-schema-registry-client</artifactId>
        </exclusion>
    </exclusions>
</dependency>

2. 配置Kafka生产者原生序列化 #

在application.yml中指定Kafka生产者的序列化器，禁用Schema相关转换逻辑：

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092 # 替换为你的Kafka集群地址
        bindings:
          # "output"对应发送消息时使用的绑定名称
          output:
            producer:
              use-native-encoding: true # 启用原生编码，绕过Stream内置消息转换器
              configuration:
                key.serializer: org.apache.kafka.common.serialization.StringSerializer
                value.serializer: org.apache.kafka.common.serialization.StringSerializer
      bindings:
        output:
          destination: your-target-topic # 替换为目标Kafka主题名

如果需要发送JSON格式的自定义对象，可将value.serializer替换为org.springframework.kafka.support.serializer.JsonSerializer，同样无需依赖Schema Registry。

3. 自定义StreamBridge Bean（可选但推荐） #

如果默认StreamBridge仍未生效，可手动创建自定义Bean，明确指定无Schema依赖的生产者配置：

import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;

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

import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;

@Configuration
public class StreamBridgeConfig {

    @Bean
    public StreamBridge streamBridge(BinderFactory binderFactory) {
        // 自定义生产者配置
        Map<String, Object> producerConfigs = new HashMap<>();
        producerConfigs.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        producerConfigs.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        
        // 创建无Schema依赖的生产者工厂与KafkaTemplate
        DefaultKafkaProducerFactory<Object, Object> producerFactory = new DefaultKafkaProducerFactory<>(producerConfigs);
        KafkaTemplate<Object, Object> kafkaTemplate = new KafkaTemplate<>(producerFactory);
        
        // 初始化自定义StreamBridge
        return new StreamBridge(binderFactory, kafkaTemplate);
    }
}

验证使用 #

在业务代码中注入自定义的StreamBridge即可发送消息：

import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageSender {

    private final StreamBridge streamBridge;

    public KafkaMessageSender(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    public void sendMessage(String message) {
        // 第一个参数对应配置中的绑定名称"output"
        streamBridge.send("output", message);
    }
}

内容的提问来源于stack exchange，提问作者mrcornel