[Kafka] Serializer, Deserializer

kafka message queue를 service 사이의 메시지 전송의 매개체로 사용하는 경우

각 서비스들에서 메시지를 produce, consume 하는 dto class가 다를 수 있다.

 

이러한 경우에 어떻게 spring-kafka 설정을 가져가야 하는지 알아보도록 하자.

 

기본 kafka 설정 

spring:
  kafka:
    producer:
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer

 

보통은 위와 같이 어떠한 종류의 serializer, deserializer를 사용할 것인지 설정해 준다.

JsonSerializer를 사용하면 모든 Java 객체를 JSON 바이트[]로 작성할 수 있습니다. 

JsonDeserializer는 byte []를 적절한 대상 객체로 역직렬화할 수 있도록 추가 Class <?> targetType 인수가 필요합니다.

 

다음 예에서는 JsonDeserializer를 만드는 방법을 보여줍니다.

JsonDeserializer<Thing> thingDeserializer = new JsonDeserializer<>(Thing.class);

 

이러한 추가적으로 필요한 class 정보를 어떻게 전달할 수 있는지 몇 가지 방법을 살펴보자.

 

JSON - Mapping Types

spring:
  kafka:
    producer:
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        spring.json.type.mapping: "my-message:com.test.Message"
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.type.mapping: "my-message:com.consumer.test.MyMessage"

spring.json.type.mapping 설정을 통해서 consumer, producer 사이의 타입 정보를 명시적으로 기입해 준다. 

장점 : 설정만 추가해 주고 관련해서 작성할 코드가 없다

단점: consumer, producer 모두 설정 필요 

 

Delegating Deserizlizer - By Topic

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.DelegatingByTopicDeserializer
      properties:
        spring.kafka.value.serialization.bytopic.config: >
          my-topic1:com.test.kafka.serialization.TestDeserializer

@Component
@Slf4j
public class TestDeserializer implements Deserializer {

    private final ObjectMapper objectMapper;
	// ...
}

 

consumer의 deserializer를 DelegatingByTopicDeserializer로 설정해 주고

어떠한 토픽에 어떠한 deserializer를 사용할 것인지 설정 정보에 매핑해 준다.

매핑한 정보에 해당하는 deserializer가 custom class인 경우에는 직접 deserializer를 imple 받은 class를 작성해야 한다.

장점 : consumer에서만 설정

단점 : topic별 deserializer class가 필요할 수도 있음 

 

관련 reference : https://docs.spring.io/spring-kafka/reference/kafka/serdes.html