一文详解kafka序列化器和拦截器

//序列化
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", StringSerializer.class.getName());

props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());

/**
 * 功能说明： JSON序列化
 * <p>
 * Original @Author: steakliu , 2022-11-02  15:14
 */
public class JsonSerializer<T> implements Serializer<T> {

  @Override
  public byte[] serialize(String topic, T obj) {
    try {
      return obj == null ? null : JSON.toJSONBytes(obj);
    }catch (Exception e){
      throw new SerializationException("json serializing exception");
    }
  }
  
}

/**
 * 功能说明：JSON反序列化
 * <p>
 * Original @Author: steakliu-刘牌, 2022-11-11  09:38
 */
public class  JsonDeserializer<T> implements Deserializer<T> {
  @Override
  public T deserialize(String topic, byte[] data) {
    return (T) JSON.parse(data);
  }
}

/**
 * 功能说明： 消费者拦截器
 * <p>
 * Original @Author: steakliu-刘牌, 2023-03-15  10:17
 */
public class MyConsumerInterceptor implements ConsumerInterceptor<String, Message> {

  @Override
  public ConsumerRecords<String, Message> onConsume(ConsumerRecords<String, Message> records) {
    long currentTimeMillis = System.currentTimeMillis();
    records.forEach(record -> {
      if ("消息异常".equals(record.value().getMessageText())) {
        //处理异常消息
        this.handleMsg(record);
      }
    });
    return records;
  }
  
  private void handleMsg(ConsumerRecord<String, Message> record) {
    //处理异常消息
  }
  @Override
  public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {}
  @Override
  public void close() {}
  @Override
  public void configure(Map<String, ?> configs) { }
}

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    consumer:
      # 反序列化器
      key-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        # 拦截器
        interceptor:
          classes: com.steakliu.kafka.interceptor.MyConsumerInterceptor
        spring:
          json:
            trusted:
              packages: '*'
    producer:
      key-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        # 拦截器
        interceptor:
          classes: com.steakliu.kafka.interceptor.MyProducerInterceptor,com.steakliu.kafka.interceptor.MyProducerInterceptor2