问如何配置Spring cloud stream (kafka)使用protobuf作为序列化

EN

public class ProtobufMessageConverter<T extends AbstractMessage > extends AbstractMessageConverter
{
    private  Parser<T> parser;
    public ProtobufMessageConverter(Parser<T> parser)

    {
        super(new MimeType("application", "protobuf"));
        this.parser = parser;
    }

    @Override
    protected boolean supports(Class<?> clazz)
    {
        if (clazz != null)
        {
            return EquipmentProto.Equipment.class.isAssignableFrom(clazz);
        }
        return true;
    }

    @Override
    public Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint)
    {
        if (!(message.getPayload() instanceof byte[]))
        {
            return null;
        }
        try
        {
//            return EquipmentProto.Equipment.parseFrom((byte[]) message.getPayload());
            return parser.parseFrom((byte[]) message.getPayload());
        }
        catch (Exception e)
        {
            this.logger.error(e.getMessage(), e);
        }
        return null;
    }

    @Override
    protected Object convertToInternal(Object payload, MessageHeaders headers, Object conversionHint)
    {
        return ((AbstractMessage) payload).toByteArray();
    }

}