[Spring MSA] Spring Event, SNS, SQS를 사용하여 DB 정합성 보장하기 1탄

나는 MSA 프로젝트에서 DB 정합성 보장을 위해 Spring Event를 다음과 같이 사용했다.

 

잠시 이전 글의 서론을 다시 가져왔다. 한번 읽어보자

내가 구성한 MSA 프로젝트에서는 멤버 서버와 또 다른 서버가 상호 작용한다. 이때, 유저가 닉네임을 변경하면 멤버 서버의 닉네임 변경 메서드는 jpa를 통해 데이터베이스에 닉네임 변경 사항을 변경감지로 엔티티의 상태를 변경한다. 이후 스프링 이벤트를 발행해서 이걸 구독하는 구독자를 2개로 나눠서 한 구독자는 AWS의 SNS에 이벤트를 발행하고 다른 구독자는 멤버 서버의 DB에 SNS이벤트를 발행했다는 기록을 남기도록 했다. (boolean사용) 이 내용은 전부 하단의 2022 우아콘에서 권용근 연사님께서 발표해 주신 내용을 토대로 구성해 본 것이다. 너무나도 감사한 마음을 가지며 따라해 보았다. 이에 이 내용을 공유한다. (영상은 문제가 된다면 바로 내리도록 하겠습니다.)

 

 

https://www.youtube.com/watch?v=b65zIH7sDug

권용근 연사님께서 발표해주신 2022우아콘 발표

 

 

1. 나의 MSA 프로젝트 구성 설명

---

1-1. 위의 영상의 방식을 사용하게 된 계기

나도 SNS, SQS를 통해서 MSA를 구성하던 중이었는데 이때 각 프로젝트의 DB 정합성을 어떻게 해결할지 구상하고 정보를 알아보면서 고민하던 도중 위의 우아콘 영상을 발견하게 되었고 이 영상을 수도없이 반복해서 보며 나름 필요한 부분들만 뽑아내서 내 프로젝트에 적용시켜 봤다.

Spring Event 아키텍처

 

1-2. 코드 구성 설명

• 위의 이미지에 붉은 원이 있는 부분은 Transaction으로 하나로 묶어놓은 코드이다. 만약 유저가 닉네임 변경을 하게되면 1번의 비즈니스 로직이 실행되면서 멤버서버의 DB에 변경된 닉네임이 저장되고 스프링 이벤트를 발행한다.
  
  
• 스프링 이벤트 리스너(구독자)는 2개를 작성했고 이 리스너들은 스프링 이벤트가 발행되면 둘 다 동작하게 되는데 이때 한 리스너(구독자)는 멤버 DB에 이벤트를 발행했다는 것을 기록으로 남기고 다른 리스너(구독자)는 AWS의 SNS에 닉네임이 변경되었다는 이벤트를 발행하게 된다. 
• 중요한 건 이 상황에 내가 빨간 원으로 그려놓은 부분(DB에 저장하는 부분)은 하나의 트랜잭션으로 묶여있다는 사실이다. 원안에 들어있지 않은 SNS 이벤트를 발행하는 스프링 이벤트 리스너(구독자)는 아래의 원 내부의 코드가 이상 없이 동작하여 트랜잭션이 완료되어 DB에 내용이 commit 되면 동작하도록 이벤트 리스너 어노테이션에 AFTER_COMMIT을 설정해 주었다.
  
  
• 이렇게 해서 이벤트 발행에 대한 정합성을 보장하도록 설계했다. 영상에 따르면 배치서버에서 5분마다 멤버의 이벤트 Record 데이터를 저장한 table을 확인해서 발행에 실패한 이벤트는 다시 발행하도록 했다.(SNS, SQS가 재발행을 해주는 최대 시간이 5분이라고 들었다.)
  
• 나는 아직 배치서버까지는 만들지 않았지만 그 부분을 제외한 내가 정리한 부분을 앞으로 설명하도록 하겠다.

 

 

2. 유저가 닉네임을 변경한다 (Controller)

---

2-1. 유저가 닉네임을 변경한다. (controller 호출)

• 이 코드에서 원래는 Param으로 Request객체를 받아야 하지만 예시를 위해 코드는 만들지 않았다. (추후 MemberRequest라는 객체를 Param으로 받아서 이 데이터를 가지고 비즈니스 로직에 보내서 이벤트를 동작시키도록 수정할 것이다. 지금의 코드는 이벤트 발행만 검증하기 위한 완전 초기 테스트 단계라 제대로 된 로직은 작성하지 않고 기본적인 동작 테스트만 진행했다.)

@Slf4j
@RequiredArgsConstructor
@RequestMapping("/member")
@RestController
public class MemberController {

    private final MemberService memberService;

    @PostMapping("/nicknameChange")
    public String nicknameChange() {

        memberService.nicknameChange();
        return "success";
    }

}

 

 

3. 컨트롤러는 비즈니스 로직을 호출한다. (Service)

---

3-1. MemberService의 비즈니스 로직 설명

• 멤버 서비스 클래스에는 유저의 닉네임을 변경시켜주는 nicknameChange() 메서드를 작성했다.
• 이 메서드 안에서 change메서드를 통해 JPA의 변경감지 기능을 활용하여 DB에 닉네임 변경(업데이트)을 하고 스프링 이벤트를 발행한다. (여기서 중요한건 JPA의 동작에 의해 닉네임 변경은 바로 DB에 커밋되지 않고 스프링 이벤트를 발행한다. 이후 그 이벤트를 구독하는 구독자도 트랜잭션을 가졌다면 구독자의 로직까지 모두 성공적으로 완료되어 트랜잭션이 끝나면 DB에 모든 내용이 커밋되어 닉네임이 업데이트된다.)

스프링 이벤트를 발행하는 Service로직

@Transactional(readOnly = true)
@Slf4j
@Service
@RequiredArgsConstructor
public class MemberService {

    private final MemberRepository memberRepository;
    private final ApplicationEventPublisher eventPublisher;

    @Transactional
    public void nicknameChage() {
        Member member = memberRepository.findById(2L).orElseThrow(() -> new MemberApplicationException(ErrorCode.DB_ERROR));
        member.changeNickname("NEW-NICKNAME222");

        log.info("닉네임 변경 Service [멤버 pk : {} , 변경전 닉네임 : {} , 변경할 닉네임 : {}]", member.getId(), "oldNickname", "NEW-NICKNAME222");
        eventPublisher.publishEvent(new NicknameChangeEvent(member.getId()));
    }

}

• 발행하고자 하는 스프링 이벤트 코드는 다음과 같다. (매우 간단하게 작성했다. Record타입으로 선언하였으며 필드값으로는 memberId만 가지도록 했다. 추후 ZeroPayload를 활용할 것이기 때문에 이벤트에는 많은 필드가 필요 없다.)

package com.recipia.member.event;

/**
 * 닉네임 변경 이벤트 객체
 * @param memberId member pk
 */
public record NicknameChangeEvent(
        Long memberId
) {
}

 

3-2. 스프링 이벤트가 발행 후 동작할 이벤트 리스너

• 닉네임 변경 스프링 이벤트가 발행되었을 때는 아래와 같이 2가지의 리스너(구독자) 메서드가 동작하도록 프로젝트를 설계하였다. 아래의 이미지와 같이 2개의 Spring Event Listener(구독자) 메서드가 동작할 것이다.

Spring Event 발행

 

 

이제 SNS의 이벤트 발행 여부를 저장하는 스프링 이벤트 리스너(구독자)에 대해서 알아보자

 

4. 비즈니스 로직에 의해 SNS 이벤트 발행여부를 저장하는 스프링 이벤트 리스너(구독자)가 동작한다.

---

4-1. 스프링 이벤트 리스너(구독자)의 동작 알아보기

• 스프링 이벤트의 기본적인 동작 원리를 지금 내 프로젝트에 빗대어 설명하자면 닉네임 변경(NicknameChangeEvent) 이벤트를 구독하고 있는 스프링 이벤트 리스너는 여러 개가 존재할 수 있는데 이때  닉네임 변경(NicknameChangeEvent) 이벤트가 발행되면 이 리스너들은 전부 매핑되어서 코드가 동작한다. (참고로 @EventListener를 메서드 상단에 적어둬야만 이벤트가 매핑된다.)

위의 설계를 보면 알 수 있듯이 나는 한 개의 NicknameChangeEvent 이벤트가 발행되었을 때 2개의 이벤트 리스너 코드가 동작하도록 하기 위해 NicknameChangeEvent를 구독하는 2개의 리스너를 만들었다. 지금부터 설명할 첫 번째 이벤트 리스너(구독자) 코드는 Member DB에 발행된 이벤트 정보를 기록(저장)한다. (이때 SNS발행을 담당하는 스프링 이벤트 리스너와는 다르게 DB에 SNS이벤트 발행 정보를 저장하는 이 이벤트 리스너는 트랜잭션을 이 이벤트를 발행한 서비스 코드와 묶어주었다.)

 

4-2. 스프링 이벤트 리스너의 트랜잭션(@Transactional) 설명

1. 나는 Spring 이벤트 리스너 메서드의 상단에 @Transactional과 @EventListener를 같이 적어줬다. 그러면 이 스프링 이벤트를 발행한 서비스 계층의 메서드인 nicknameChange()의 트랜잭션(@Transactional)과 지금 이 이벤트 리스너 메서드의 트랜잭션(@Transactional)이 연결되어 하나의 트랜잭션으로 묶이게 된다.
   
   
2. 이렇게 하나의 트랜잭션으로 묶어주게 되었을 때 얻을 수 있는 장점은 다음과 같다. 먼저 유저의 닉네임 변경이 발생했을 때의 동작을 보면 다음과 같다.

1. 멤버 DB에 실제로 변경된 닉네임을 저장한다.
2. 스프링 이벤트를 발행해서 AWS SNS에 내가 닉네임 변경 이벤트를 발행했다는 기록을 멤버 DB안에 저장한다.

• 이렇게 2가지의 다른 동작을 한 트랜잭션으로 묶어서 처리할 수 있기 때문에 다음과 같은 장점을 얻을 수 있다.

1. 만약 닉네임 변경 비즈니스 로직이 동작하던 도중 오류가 발생한다면 애초에 이벤트는 발행되지 않을 것이다.(트랜잭션에 의해 롤백이 되니 그렇다. 참고로 SNS발행 이벤트도 AFTER_COMMIT이라 커밋이 성공했을 때만 동작한다.)

2. 그게 아니라 이벤트는 잘 발행되었지만 SNS를 전송하는 요청인 HTTP에 문제가 생겨서 SNS발행이 실패했다고 하더라도 이때는 멤버 DB에 남아있는 SNS 발행여부가 기록으로 남아있기 때문에 배치서버를 통해서 확인해서 재발행을 할 수 있게 된다.

 

4-3. 이벤트 기록을 저장할 DB Table 설명

• 나는 Record 테이블을 아래와 같이 구성했다. 위 영상의 권용근 연사님께서 알려주신 것과 비슷하지만 조금 다른 부분들이 있는데 그 이유는 나는 이 프로젝트를 진행하면서 팀원과 많은 토의를 하는데 이때 우리의 프로젝트에서 필요로 하는 속성을 같이 찾고 고민해서 그것을 지정하여 테이블을 생성했다. 만약 이 내용을 보고 바로 따라 해보고자 하는 분이 있다면 그러지 말고 많은 고민과 생각을 해보고 필요한 부분만을 가지고 가서 본인의 프로젝트에서 사용하는 것을 추천한다.

member_event_record 테이블

 

4-4. 실제 이벤트 리스너(구독자) 코드 설명

• 아래의 코드는 DB에 변경된 닉네임을 저장하고 Spring event를 발행하는 서비스 계층의 메서드다.

스프링 이벤트를 발행하는 Service로직

 

• 아래의 코드가 바로 위의 서비스 코드에서 닉네임 변경 스프링 이벤트가 발행되면 동작하는 스프링 이벤트 리스너(구독자) 메서드다. 잘 보면 메서드 상단에 @Transactional이 작성되어 있는 것을 볼 수 있다. (이렇게 서비스랑 리스너를 하나의 트랜잭션으로 묶어줬다.)

스프링 이벤트 발행여부 저장하는 구독자

 

• 위의 코드를 보면 알 수 있겠지만 처음 DB에 이벤트 발행 여부를 기록할 때 published는 false로 저장한다. 이후 SNS -> SQS까지 잘 전달되면 SQS가 이 published 컬럼을 true로 변경시켜 줄 것이다.(이런 식으로 이벤트가 잘 보내졌는지를 검증한다.)

Record 테이블에 published false로 저장

@RequiredArgsConstructor
@Component
public class EventRecordListener {

    private final MemberRepository memberRepository;
    private final MemberEventRecordRepository memberEventRecordRepository;
    private final AwsSnsConfig awsSnsConfig;
    private final CustomJsonBuilder customJsonBuilder;

    /**
     * 이벤트를 호출한 서비스 코드의 트랜잭션과 묶이게 된다.
     */
    @Transactional
    @EventListener
    public void listen(NicknameChangeEvent event) throws JsonProcessingException {
        // 여기서 db에 저장하는 로직 실행 (트랜잭션이 묶여있어야 함)
        Member member = memberRepository.findById(event.memberId()).orElseThrow(() -> new MemberApplicationException(ErrorCode.USER_NOT_FOUND));

        // JSON 객체 생성 및 문자열 변환
        String messageJson = customJsonBuilder.add("memberId", member.getId().toString()).build();

        // SNS 토픽명 분리
        String topicName = MemberStringUtils.extractLastPart(awsSnsConfig.getSnsTopicNicknameChangeARN());

        MemberEventRecord memberEventRecord = MemberEventRecord.of(
                member,
                topicName,
                "NicknameChangeEvent",
                messageJson,
                false,
                null
        );

        memberEventRecordRepository.save(memberEventRecord);
    }

}

 

4-5. 스프링 이벤트 리스너에서 사용하는 Util 클래스 1: CustomJsonBuilder

• 아래의 CustomJsonBuilder 클래스는 이벤트 기록(Member_Event_Record) 테이블이 가진 컬럼중에 attribute(메시지 정보)에 값을 Json으로 넣어주기 위해 변환시켜 주는 유틸 클래스다.

이벤트 기록 테이블의 attribute 컬럼에 memberId를 Json으로 key:value 형태로 값을 넣어줘서 추후 이벤트 발행이 실패하였을 때 DB를 통해서 attribute 정보를 확인한 후 다시 같은 이벤트를 발행할 때 attribute 컬럼에서 바로 key값으로 value에 넣어준 memberId를 꺼내서 재발행에 사용하기 위해서다. 이렇게 설계하면 추후 배치를 통해 다시 이벤트를 발행하기가 쉬워진다.

 

• 리스너 로직에서 아래와 같이 CustomJsonBuilder 클래스를 통해서 memberId를 json으로 key, value형태로 만들어 줬다.

json 객체 생성

@Component
public class CustomJsonBuilder {

    private Map<String, Object> values = new HashMap<>();

    public CustomJsonBuilder add(String key, Object value) {
        values.put(key, value);
        return this;
    }

    public String build() throws JsonProcessingException {
        ObjectMapper objectMapper = new ObjectMapper();
        return objectMapper.writeValueAsString(values);
    }

}

 

4-6. 스프링 이벤트 리스너에서 사용하는 Util 클래스 2: MemberStringUtils

• 아래의 MemberStringUtils 클래스는 SNS ARN 값을 받아온 후에 모든 값을 그대로 넣는 것이 아니라 맨 뒤에 적히게 될 topic이름만 분리해서 저장하도록 도와주는 String 유틸 클래스이다.
• 아래의 ARN을 예시로 하자면 맨 뒤의 member-test만 저장하도록 하는 것이다. (이미 삭제한 SNS라서 모자이크 처리 X)

arn 예시

 

• 실제 리스너 로직에서는 아래와 같이 arn의 ":" 를 기준으로 맨 뒤의 topic 이름만 가져오도록 하는 데 사용되었다.

SNS arn에서 topic이름만 분리하기

/**
 * 이 유틸 클래스를 만든 이유는 MemberEventRecord에서 SNS토픽의 arn을 그대로 들고오는데
 * 이건 보안이기 때문에 ":"를 기준으로 맨 뒤의 SNS 토픽명만 가져오도록 만든 것이다.
 */
public class MemberStringUtils {

    public static String extractLastPart(String input) {
        int lastIndex = input.lastIndexOf(":");
        if (lastIndex != -1) {
            return input.substring(lastIndex + 1);
        } else {
            return "";  // ':'이 없으면 빈 문자열 반환
        }
    }
}

 

다음으로 AWS SNS에 실제로 이벤트를 발행하는 스프링 이벤트 리스너(구독자)를 알아보자

 

5. 비즈니스 로직에 의해 실제로 AWS의 SNS에 이벤트를 발행하는 스프링 이벤트 리스너(구독자)도 동작한다.

---

5-1. 아래의 이벤트 리스너 코드는 AWS의 SNS에 닉네임 변경 이벤트를 발행하는 코드다.

• 이 이벤트 리스너는 메서드 상단에 @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)를 작성했기 때문에 이 이벤트를 호출한 서비스의 트랜잭션이 commit이 되어 종료가 된 후에 이 이벤트 리스너가 동작하여 AWS의 SNS에 닉네임을 변경했다는 이벤트를 발행하게 된다.

AWS의 SNS에 이벤트를 발행하는 스프링 이벤트 리스너(구독자)

@Slf4j
@RequiredArgsConstructor
@Component
public class SnsPublishListener {

    private final SnsService snsService;
    private final CustomJsonBuilder customJsonBuilder;

    /**
     * 이벤트를 호출한 서비스 코드의 트랜잭션과 묶여있지 않고 트랜잭션이 commit된 후에 동작한다.
     */
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void snsListen(NicknameChangeEvent event) throws JsonProcessingException {

        String messageJson = customJsonBuilder.add("memberId", event.memberId().toString()).build();
        snsService.publishNicknameToTopic(messageJson);
    }

}

 

 

5-2. @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)

• 위의 어노테이션을 사용함으로써 이 이벤트 구독자는 이벤트를 호출한 발행자의 트랜잭션이 종료되었을 때 코드가 동작하도록 설정할 수 있게 되었다. 이것에 대한 내용은 springframework4.2부터 적용되었는데 그 내용은 아래의 글을 참고하면 좋을 것이다.

 

[Spring] SpringFramework 4.2 이후 스프링 이벤트의 변화

 

 

6.  SNS의 이벤트 메시지 발행과 이에 따른 SQS의 동작 설명

---

6-1. 이벤트의 SNS 메시지 발행

• 스프링 이벤트 리스너(구독자)는 위의 과정들을 거쳐서 AWS의 SNS에 닉네임이 변경되었다는 메시지를 발행하였다. (하단의 postman으로 요청을 보낸 후에 스프링 이벤트 발행 후 커밋까지 하고 나서 AWS의 SNS메시지 발행하는 것을 확인했다.)

postman으로 sns 발행 확인하기

 

• 아래와 같이 SNS에 메시지를 발행했다는 로깅이 남았다.

SNS 발행 로그 확인하기

 

6-2. SQS의 동작

• SNS로부터 메시지를 받은 SQS는 여러 개가 존재할 텐데(다른 서버들) 그중 한 개의 SQS는 멤버 DB의 이벤트 발행 여부를 저장하는(MEMBER_EVENT_RECOED) 테이블 전용으로 만들어서 만약 이벤트를 잘 받았다면 멤버 DB의 SNS이벤트 발행여부(MEMBER_EVENT_RECOED) 테이블의 published컬럼값을 false에서 true로 변경시켜 준다.(이벤트가 잘 실행되었다고 DB상태를 변경시켜 주는 것이다.)

SQS의 동작 설명

 

 

7. 데이터 확인

---

7-1. Data 확인하기

• 마지막으로 나는 위의 과정들을 다 거친 후에 DataGrip에 들어가서 내가 작성한 코드들이 제대로 동작했는지를 확인했다. 아래 이미지를 보다시피 문제없이 잘 동작한 것을 확인했다.

데이터 확인

 

 

8. 느낀 점

---

이 모든 과정을 실제로 경험해 보면서 이것들에 대한 확실한 검증을 해보기 위해서는 실제로 SNS 요청 발생 시에 HTTP 관련 문제가 발생했을 때 내가 직접 그 상황을 경험해 봐야 할 것이고(일부로 발생시켜 보도록 할 것이다.) 또한 발행에 실패한 메시지에 대해서는 5분마다 배치를 돌리는 작업을 하는 코드를 직접 작성하고 동작시켜 봐야 확실하게 이 코드에 문제가 없는지 알 수 있을 것이다. 다만 지금까지 해본 과정을 통해서도 나는 굉장히 많은 것을 배웠다고 생각한다.
나는 이번 코딩을 통해 스프링 이벤트에 대해서 배웠으며 HTTP요청이 있을 때 외부 이벤트를 발행한다면 이때는 어떻게 이벤트가 발행되는 것을 보장해 줄지도 배웠으며 MSA에서 데이터의 정합성을 맞춰가는 방법까지 배웠다.
아직 해볼 것이 산더미처럼 많은 상황이지만 이렇게 한 발자국 앞으로 내딛을 수 있게 도와주신 우아콘 2022년 연사님이신 권용근 연사님께 엄청난 감사함을 느낀다. 많이 힘들었는데 덕분에 이렇게 MSA 구축에 한 발자국을 내딛을 수 있게 되었다. 다음에는 이 기능을 확실히 구현하여 보충 설명을 하는 포스트로 돌아오도록 하겠다.

 

 

 

 

이어지는 내용은 아래의 포스트를 읽어보도록 하자 (ZeroPayload를 적용한 FeignClient요청이 이루어 진다.)

Spring Event, SNS, SQS를 활용한 MSA 데이터 정합성 보장 전략 2탄 - ZeroPayload 방식으로 FeignClient 요청하

Message-Driven이 뭔지 아래의 글을 보는게 도움이 될 것이다.

AWS Message-Driven 입문 1편: SNS 설정으로 시작하는 MSA