如何使用org.springframework.messaging.converter.ProtobufMessageConverter

开发者阿泽

2026-9-24

问题根因

Spring Cloud Stream 默认不会将ProtobufMessageConverter加入默认转换器列表，且ContentTypeConfiguration.isConverterEligible的过滤逻辑默认仅放行内置白名单转换器，手动实例化的ProtobufMessageConverter默认不会被纳入转换链路。

解决方案

步骤1：添加必要配置项

在application.yml中添加配置，将ProtobufMessageConverter加入允许的转换器白名单，避免被过滤：

spring:
  cloud:
    stream:
      # 将Protobuf转换器加入可生效的转换器列表
      content-type-converters: org.springframework.http.converter.protobuf.ProtobufMessageConverter
      bindings:
        input-in-0:
          destination: test-input
          group: test-group
          content-type: application/x-protobuf;charset=UTF-8

步骤2：手动注册高优先级的Protobuf转换器

自定义配置类，实例化ProtobufMessageConverter并设置高优先级，确保它在默认转换器前执行：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.http.MediaType;
import org.springframework.http.converter.protobuf.ProtobufMessageConverter;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import com.google.protobuf.ExtensionRegistry;
import java.util.ArrayList;
import java.util.List;

@Configuration
public class ProtobufConverterConfig {

    @Bean
    @Order(Ordered.HIGHEST_PRECEDENCE)
    public CompositeMessageConverter protobufCompositeConverter() {
        List<MessageConverter> converterList = new ArrayList<>();

        // 初始化Protobuf转换器，若用到Protobuf扩展可自定义ExtensionRegistry
        ProtobufMessageConverter protobufConverter = new ProtobufMessageConverter(
            ExtensionRegistry.newInstance()
                // 此处可注册你用到的Protobuf扩展，无扩展可直接用无参构造
                // .add(YourProtoClass.yourExtension)
                .build()
        );
        // 显式声明支持的媒体类型，确保和绑定配置的content-type匹配
        protobufConverter.setSupportedMediaTypes(List.of(
            MediaType.parseMediaType("application/x-protobuf"),
            MediaType.parseMediaType("application/x-protobuf;charset=UTF-8")
        ));
        converterList.add(protobufConverter);

        // 加入默认转换器，不影响其他格式消息的处理逻辑
        converterList.addAll(new CompositeMessageConverterFactory().initDefaultConverters());

        return new CompositeMessageConverter(converterList);
    }
}

步骤3：函数式消费端参数适配

确保你的消费函数参数直接声明为Protobuf生成的类，框架会自动匹配转换：

import org.springframework.context.annotation.Bean;
import java.util.function.Consumer;

// YourProtoClass为protoc生成的Protobuf消息类
@Bean
public Consumer<YourProtoClass> inputIn0() {
    return protoMsg -> {
        // 直接处理转换完成的Protobuf对象
        System.out.println("收到消息：" + protoMsg.toString());
    };
}

注意事项

确保项目已引入com.google.protobuf:protobuf-java依赖，版本需和你protoc编译用的Protobuf版本一致，避免兼容性问题。
若仍不生效可排查转换器优先级，确认ProtobufMessageConverter在处理二进制的默认转换器（比如ByteArrayMessageConverter）之前执行，避免消息被提前转换为字节数组导致链路中断。
生产端发送Protobuf消息时，对应output binding也需配置content-type: application/x-protobuf，转换器会自动序列化消息。

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