使用MessageBuilder和KafkaTemplate发Kafka消息前如何移除payload空字段

方案1：配置JSON序列化规则（推荐，侵入性最低） #

如果你的payload是通过Jackson序列化为JSON字符串发送的，直接配置Kafka的序列化器属性即可自动过滤空字段，无需修改业务代码：

• 如果你使用Spring Kafka，直接在application.yml/application.properties中添加配置：

spring:
  kafka:
    producer:
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        # 仅过滤null值用NON_NULL，同时过滤null、空字符串、空集合用NON_EMPTY
        spring.json.serialization.inclusion: NON_EMPTY

• 如果你是手动构造JsonSerializer实例，直接通过构造参数配置：

ObjectMapper objectMapper = new ObjectMapper()
  .setSerializationInclusion(JsonInclude.Include.NON_EMPTY);
JsonSerializer<YourPayloadClass> serializer = new JsonSerializer<>(objectMapper);

该方案会在序列化阶段自动移除所有符合规则的空字段，MessageBuilder构建逻辑完全不需要修改。

方案2：构建Message前预处理payload #

如果你需要更灵活的过滤规则，或使用的不是JSON序列化，可以在调用MessageBuilder前先处理payload对象：

• 对于Map类型的payload：

Map<String, Object> payload = new HashMap<>();
// 填充payload字段后执行过滤
payload.entrySet().removeIf(entry -> 
  entry.getValue() == null 
  || (entry.getValue() instanceof String && ((String) entry.getValue()).trim().isEmpty())
);
Message<Map<String, Object>> message = MessageBuilder.withPayload(payload).build();

• 对于自定义Java Bean类型的payload，可以通过反射遍历字段过滤，或者用Jackson先转成过滤后的Map再使用：

ObjectMapper objectMapper = new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_EMPTY);
Map filteredMap = objectMapper.convertValue(yourBean, Map.class);
Message<Map> message = MessageBuilder.withPayload(filteredMap).build();

方案3：通过KafkaTemplate全局拦截器统一处理 #

如果需要对所有发送的消息都生效，不需要逐个修改业务发送逻辑，可以实现ProducerInterceptor统一处理：

public class EmptyFieldFilterInterceptor implements ProducerInterceptor<Object, Object> {
  private final ObjectMapper objectMapper = new ObjectMapper()
    .setSerializationInclusion(JsonInclude.Include.NON_EMPTY);

  @Override
  public ProducerRecord<Object, Object> onSend(ProducerRecord<Object, Object> record) {
    Object payload = record.value();
    // 过滤处理payload逻辑，这里以转成过滤后的JSON字符串为例
    try {
      String filteredPayload = objectMapper.writeValueAsString(payload);
      return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(),
        record.key(), filteredPayload, record.headers());
    } catch (JsonProcessingException e) {
      throw new RuntimeException("payload过滤失败", e);
    }
  }

  @Override
  public void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
  @Override
  public void close() {}
  @Override
  public void configure(Map<String, ?> configs) {}
}

然后将拦截器配置到Kafka生产者属性中：

spring:
  kafka:
    producer:
      properties:
        interceptor.classes: com.yourpackage.EmptyFieldFilterInterceptor

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