이번 포스팅에서는 MSA 전환 후 채팅 마이크로 서비스 개발 하면서 겪었던 트러블 슈팅 두가지에 대해 정리 해볼려고 합니다.
👉 참고 코드 : https://github.com/LminWoo99/PlantBackend/tree/msa-master
JPA open-in-view SSE Connection 고갈 문제
채팅 마이크로 서비스를 개발하면서 SSE를 활용한 알림 기능을 구현했었습니다. 보통 SSE 통신을 하는 동안은 HTTP Connection이 계속 열려있습니다. 만약 SSE 연결 응답 API에서 JPA를 사용하고 open-in-view 속성이 true로 되어있다면, HTTP Connection이 열려있는 동안 DB Connection도 같이 열려있게 됩니다.
위 사진은 open-in-view 설정이 true 일때의 transaction 로그입니다
2번째 로그를 보면 JPA transaction이 commit 되었지만, open-in-view 속성때문에, JPA EntityManager는 닫히지 않습니다. 이때문에 트랜잭션이 끝나도 DB connection도 반납되지 않았습니다.
즉 DB Connection Pool에서 최대 10개의 Connection을 사용할 수 있다면, 10명의 클라이언트가 SSE 연결 요청을 하는 순간 DB 커넥션도 고갈되게 됩니다. 따라서 이 경우 open-in-view 설정을 반드시 false로 설정해야 합니다.
단순히 이런 문제점만 생각하고 false 처리를 해버리면 DB Connection이 해제되지 않는 문제는 해결될 것입니다. 하지만 그럴 경우에는 lazy loading 을 사용할 때 문제가 발생할 수 있습니다. EntityManger 가 transaction의 commit과 함께 닫히기 때문에, 트랜잭션 외부에서 lazy loading 을 사용하면 no Session 이라는 예외가 발생하게 됩니다. 이러한 문제점까지 고려해서 서비스의 요구사항에 맞게 문제점을 해결해야 합니다!
💡 해결
식구하자 채팅 마이크로 서비스에는 연관된 데이터와 따로 조인을 하지 않기 때문에 lazy-loading을 사용하지 않아 저는 Connection 고갈 문제를 open-in-view 설정을 false 처리하여 해결하게 되었습니다.
마이크로 서비스간 단일 DB 데이터 동기화 문제(Kafka를 통해 해결)
프로젝트를 진행하면서, MSA의 장점을 살리기 위해 서비스마다 단일 DB를 사용하고 있습니다. 현재 개발 완료된 서비스는 기존 모놀리식 서비스였던 중고 식물 거래 서비스와 채팅 서비스가 있습니다.
그래서 중고 거래 게시글에 여러개의 채팅방 채팅 내역을 가지게 됩니다. 하지만 게시글과 채팅방 및 채팅 내역들은 서로 다른 DB에 저장되어 관리하고 있습니다.
여기서 문제되는 점이 중고 식물 거래 게시글이 삭제될 경우 해당 게시글의 채팅방, 채팅 내역 데이터들도 연쇄적으로 삭제되게 구현하고 싶었습니다.
즉, 서로 다른 데이터베이스의 동기화가 필요한 상황에서 두 가지 해결 방식을 고민했습니다.
1. API 호출을 통한 동기화
한 서비스에서 다른 서비스의 API를 직접 호출하여 데이터를 동기화하는 방식입니다. 예를 들어, 게시글 삭제 API가 성공적으로 수행된 후, 채팅 서비스의 관련 데이터 삭제 API를 호출합니다.
장점: 구현이 비교적 단순하고 직관적입니다.
단점: 서비스 간의 결합도가 높아지며, 실패 시 복구 메커니즘을 따로 구현해야 합니다.
해당 방식은 구현하기는 간단하지만, 마이크로 서비스 간 최소한의 결합도를 유지해야 되기 때문에 적합하지 않다고 판단했습니다.
2. 이벤트 기반 동기화 (Event-Driven Synchronization)
Kafka를 사용하여, 한 서비스에서 발생한 이벤트(예: 게시글 삭제)를 다른 서비스가 구독하고, 해당 이벤트에 반응하여 필요한 조치(예: 채팅 방 및 내역 삭제)를 취하는 방식입니다. 이 방법은 느슨한 결합과 확장성을 제공합니다.
장점: 서비스 간의 결합도를 낮추며, 각 서비스가 독립적으로 확장될 수 있도록 합니다.
단점: 이벤트의 순서 보장, 중복 처리, 실패 시 복구 메커니즘 등을 고려해야 합니다.
서비스 간의 결합도를 낮추며, 각 서비스가 독립적으로 확장될 수 있도록하는 kafka를 통한 동기화가 적합하다고 판단하여 적용했습니다
Kafka를 통한 동기화
중고 식물 거래 서비스(plant-service)에 KafkaProducer 구현
public void deletePost(TradeBoardDto tradeBoardDto) {
tradeBoardRepository.delete(tradeBoardDto.toEntity());
/*send this deletePost to the kafka*/
kafkaProducer.send("deletePost", tradeBoardDto);
}
===================================================
@Service
@Slf4j
public class KafkaProducer {
private KafkaTemplate<String, String> kafkaTemplate;
@Autowired
public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
/**
* 중고 거래 게시글 삭제시 카프카를 통해 plant-chat-service 마이크로서비스에 전달
* @param : String topic, TradeBoardDto tradeBoardDto
*/
public TradeBoardDto send(String topic, TradeBoardDto tradeBoardDto) {
ObjectMapper mapper = new ObjectMapper();
String jsonInString = "";
try{
jsonInString = mapper.writeValueAsString(tradeBoardDto);
} catch (JsonProcessingException e) {
e.printStackTrace();
}
kafkaTemplate.send(topic, jsonInString);
log.info("Kafka Producer send data from the Plant microservice " + tradeBoardDto);
return tradeBoardDto;
}
}deletePost 메소드에서 게시글 삭제 요청이 들어오면, 해당 게시글을 데이터베이스에서 삭제하고, 이 사실을 Kafka를 통해 다른 서비스에 알립니다. 이때, kafkaProducer.send 메소드를 통해 "deletePost" 토픽으로 게시글 정보가 포함된 메시지를 전송
채팅 서비스 (plant-chat-service) Kafka Listener 등록
@Service
@Slf4j
@RequiredArgsConstructor
public class KafkaConsumer {
private final ChatService chatService;
/**
* plant-service 게시글에 속한 채팅 데이터 kafka를 통해 삭제 메서드
*
* @param : MemberDto memberDto, ChatRequestDto requestDto
*/
@KafkaListener(topics = "deletePost")
public void deleteChat(String kafkaMessage) {
log.info("Kafka Message : ->" + kafkaMessage);
Map<Object, Object> map = new HashMap<>();
ObjectMapper mapper = new ObjectMapper();
try {
map = mapper.readValue(kafkaMessage, new TypeReference<Map<Object, Object>>() {
});
//TradeBoard id값 가져오기
Integer tradeBoardNo = (Integer) map.get("id");
if (tradeBoardNo == null) {
throw new IllegalArgumentException("TradeBoard ID is missing in the Kafka message");
}
chatService.deleteChatRoom(tradeBoardNo);
} catch (JsonProcessingException e) {
log.error("Error parsing Kafka message: {}", kafkaMessage, e);
} catch (IllegalArgumentException e) {
log.error("Validation error for Kafka message: {}", kafkaMessage, e);
} catch (Exception e) {
log.error("Error processing Kafka message: {}", kafkaMessage, e);
}
}
}
===================================================
@Transactional
public void deleteChatRoom(Integer tradeBoardNo) {
List<Integer> chatRoomNoList = chatRepository.deleteChatRoomAndReturnChatNo(tradeBoardNo);
deleteChatting(chatRoomNoList);
}
@Transactional
public void deleteChatting(List<Integer> chatRoomNoList) {
// Set 자료구조를 활용하여 중복된 chatNo 제거
Set<Integer> uniqueChatNoSet = new HashSet<>(chatRoomNoList);
// 중복 제거 후의 chatNo에 해당하는 채팅 데이터 삭제
mongoTemplate.remove(query(where("chatRoomNo").in(uniqueChatNoSet)), Chatting.class);
}Kafka 메시지 수신 및 처리: @KafkaListener 어노테이션이 적용된 deleteChat 메소드는 "deletePost" 토픽으로부터 메시지를 수신합니다. 메시지는 JSON 형태로 전달되며, 이를 파싱하여 게시글 ID를 추출합니다. 이후, 해당 게시글 ID와 관련된 채팅 방을 삭제하는 로직을 실행
위와 같은 방식으로 Kafka를 통해 MSA간 단일 데이터베이스의 동기화 문제를 해결했습니다.
참고
