kafka

Spring Kafka 메시지 수신을 하는 다양한 방법

Spring Boot에서 Kafka 메시지를 수신하는 방법은 여러가지가 있다. 기본적인 방법부터 헤더정보 가져오는 방법, 여러 토픽을 한번에 받는 방법, 설정파일에 지정한 토픽이름으로 받아오는 방법 등 다양한 수신 방법에 대해서 정리한 내용이다. 1. 기본 spring-kafka를 사용하여 메시지를 수신하기 위해서는 기본적으로 @KafkaListenr를 통해 받을 수 있다. 아래 예는 message.command 토픽을 수신하는 방법이다. @Component @KafkaListener( topics = "message.command", groupId = "message.groupId" ) public class MessageSubscriber { @KafkaHandler public void handle..

2024. 3. 30. 19:07

kafka

spring kafka를 사용할 때의 트랜잭션 처리

일반적인 서비스 로직을 작성할 때 메소드 내에서 복잡한 서비스 로직을 수행하고 이벤트(kafka)를 발행하는 경우가 많다. 이런 경우 트랜잭션이 성공적으로 완료된 이후에 이벤트가 발행되기를 기대하지만 로직 내의 실행 순서에 따라 의도하지 않게 이벤트가 발행이 되는 경우도 있다. 그래서 스프링에서는 트랜잭션을 종료하고 이벤트를 발행하는 방법으로 @TransactionalEventListener를 사용을 한다. @TransactionalEventListener를 사용하게 되면 트랜잭션이 종료된 이후에 이벤트를 발행하는 것이기 때문에 이벤트 발행시점을 트랜잭션 이후로 미룰 수 있는 장점이 있다. 하지만, 여기서는 @TransactionalEventListener를 사용하지 않고 kafka 트랜잭션과 DB 트랜..

2023. 10. 10. 07:20

kafka

Spring kafka의 concurrency는 높으면 좋은걸까?

Spring Kafka를 사용할 때 아래와 같이 concucrency라는 옵션이 있다. 이 값이 높으면 성능이 좋아질까? @KafkaListener( topics = "kafka-concurrency", groupId = "kafka-concurrency-consumer-group", concurrency = "1" ) 테스트 방법 concurrency의 동작을 확인하기 위해 아래와 같은 시나리오로 테스트하였다. producer는 10개의 메시지를 동시에 보낸다. for (int i = 0; i < 10; i++) { TextMessage message = new TextMessage("message " + i); log.info("[send] " + message.getText()); kafkaTemp..

2023. 1. 20. 09:46

kafka

KafkaListener에서 서로 다른 message mapping하기

다른 2개의 서비스에서 동일 토픽을 다른 클래스(서비스내 클래스)로 매핑하는 방법이다. [A서비스에서 메시지 발행] topicId: message-topic message는 2가지 발행: service1.model.TextMessage, service1.model.FileMessage [B서비스에서 메시지 구독] topicId: message-topic message는 2가지 발행: service2.model.TextMessage, service2.model.FileMessage spring kafka에서 발행한 메시지의 헤더를 보면 __TypeID__라는 것이 있다. 이것은 메시지를 보낼 때의 데이터 객체 타입을 매핑하여 전송하게 되고 실제 수신받는 컨슈머는 이 객체타입을 사용하여 deserialize..

2022. 12. 23. 07:36

kafka

kafka publisher의 성능 비교

[궁금점] kafka publisher로 메시지 전송 시 아래와 같은 궁금점에 대해 테스트해 보았다. 동기(sync)와 비동기(async)의 성능 차이가 얼마나 날까? topic의 partition 수를 늘리면 성능이 빨라질까? producer의 acks 설정이 성능에 어떤 영향을 줄까? [테스트 환경] 개발환경: spring boot, java 8, spring-kafka kafka는 원격서버에 위치 single-kafka: brokder 1대 multi-kafka: broker 3대 (replication-factor: 3) [테스트 방법] 메시지 내용: 100개의 문자열 publisher를 1000번 실행한 시간에 대한 3회 평균 시간 [테스트 코드] // 동기 @Transactional publi..

2022. 12. 23. 07:13

kafka

kafka CommitFailedException 정리

Kafka에서 CommitFailedException이 나오는 유형은 아래와 같다. consumer 로직의 처리시간이 max.poll.interval.ms보다 클 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우 session.timeout.ms시간동안 heartbeat가 오지 않았을 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우 1. consumer 로직의 처리시간이 max.poll.interval.ms보다 클 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우 ConsumerConfig에서 @Bean public ConsumerFactory consumerFactory() { ... props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 50..

2022. 12. 23. 07:13