Spring Boot - RabbitMQ를 사용하여 알림 발송 1편

ApplicationEventPublisher란?

---

• 이벤트 리스너는 발생된 event에 반응하고 이벤트 리스너는 발생된 event를 전달받아 이벤트에 담긴 데이터를 기반으로 특정한 기능을 수행합니다.

 

🤔 왜 ApplicationEventPublisher를 사용해야할까?

---

• 아래는 회원가입 상황입니다. 클라이언트가 회원가입을 완료하면 가입 축하 알림 발송과 축하 할인 쿠폰을 발송하고 있습니다. 이는 절차지향적 관점에서는 당연한 순서이지만 몇가지 문제점을 가지고 있습니다.

@Service
@Transactional
@RequiredArgsConstructor
public class MemberService {

    private final MemberCouponService memberCouponService;
    private final NotificationService notificationService;
    private final MemberRepository memberRepository;
    

    public void create(MemberCreateDto request) {
        Member member = Member.builder()
                .name(request.getName())
                .phone(request.getPhone())
                .build();

        memberRepository.save(member); // 회원가입 완료
        notificationService.send();    // 회원가입 축하 알림 발송
        memberCouponService.send();    // 회원가입 축하 할인 쿠폰 발송
    }
}

 

🧨 문제점 1 - 강한 결합

• 회원가입시에 MemberService는 여러 책임을 동시에 수행하게 됩니다. 가입 축하 알림 발송 및 축하 할인 쿠폰 발송 책임을 담당하기 때문에 결합도가 높아집니다.

🧨 문제점 2 - 트랜잭션

• 현재 모든 로직은 하나의 트랜잭션에 참여하게 됩니다. 그렇기 때문에 알림 발송에서 예외가 발생하게 된다면 클라이언트의 회원가입 역시 롤백이 되어버립니다. 알림 발송시 알림 발송이 실패한다면 회원가입이 롤백이 되어버린다? 한번쯤 고민해볼 필요가 있습니다. 

🧨 문제점 3 - 성능

• 회원가입 완료(0.5초) 소요, 축하 알림 발송(5초) 소요, 축하 할인 쿠폰 발송(10초) 소요 = 총 15.5 초가 소요되게 됩니다. 이는 클라이언트의 관점에서 회원가입은 완료했지만 부가적인 기능들로 인해 15초 동안 렌더링 페이지를 보고만 있어야 한다는 것입니다.
  누가 이 애플리케이션을 사용할려고 할까요?

 

알림 발송을 위한 전체적인 그림

---

전체적인 구조

 

RabbitMQ 설정 적용

---

💡 yml 파일 설정

// yml 파일에 rabbitMQ 설정을 추가해줍니다.
rabbitmq:
 host: localhost
 port: 5672
 username: guest
 password: guest

 

💡 Config 설정

• 해당 예제는 서버를 Producer server, Consumer server를 나누지 않고 하나의 서버가 Producer, Consumer의 역할을 하는 예제입니다.

@EnableRabbit
@Configuration
public class RabbitMQConfig {

    @Bean
    Queue queue() {
        return QueueBuilder.durable("server-queue").build();
    }

    @Bean
    FanoutExchange exchange() {
        return new FanoutExchange("fanoutExchange");
    }
    
    @Bean
    Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange());
    }

    @Bean
    Binding deadLetterQueueBinding() {
        return BindingBuilder
                .bind(deadLetterQueue())
                .to(deadLetterExchange());
    }

    @Bean("rabbitTemplate")
    RabbitTemplate rabbitTemplate(@Qualifier("RabbitConnectionFactory") ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter());
        return rabbitTemplate;
    }

    @Primary
    @Bean("RabbitConnectionFactory")
    @ConfigurationProperties("spring.rabbitmq")
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        return connectionFactory;
    }

    @Bean
    MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

 

💡 RabbitMQ Producer 설정

• fanoutExchange로 전달하기 위해 Producer 설정을 해줍니다.

@Component
@RequiredArgsConstructor
public class RabbitMQProducer {

    private final RabbitTemplate rabbitTemplate;
    private static final String EXCHANGE_NAME = "fanoutExchange";

    @TransactionalEventListener
    public void eventHandler(NotificationMessage notificationMessage) {
        sendMessage(notificationMessage);
    }

    private void sendMessage(NotificationMessage response) {
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, "", response);
    }
}

 

💡 RabbitMQ Consumer 설정

• Consumer에서는 Queue에 담긴 메시지를 소비하는 동시에 다시 특정 Queue를 Consumer 하고 있는 클라이언트에게 메시지를 보내줍니다. 여기서 Consumer는 두가지의 일을 하게 됩니다.
  1. 알림 정보를 데이터 베이스에 보관
  2. 알림을 사용자에게 전달

@Component
@RequiredArgsConstructor
public class RabbitMQConsumer {

    private final RabbitTemplate rabbitTemplate;
    private static final String EXCHANGE_NAME = "amq.topic";
    private final ObjectMapper objectMapper;
    private final NotificationService notificationService;

    @RabbitListener(queues = "server-queue")
    public void consume(NotificationMessage notification) {
        System.out.println(notification);
        NotificationResponse response = notification.notificationSend(notificationService);
        sendMessage(response);
    }

    private void sendMessage(NotificationResponse response) {
        try {
            String json = objectMapper.writeValueAsString(response);
            rabbitTemplate.convertAndSend(EXCHANGE_NAME, response.getRoutingKey(), json);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }
}

 

알림을 발송해보자

---

• 이제 RabbitMQ에 대한 설정은 끝이 났습니다. 이제 어떻게 클라이언트에게 메시지를 보낼건가에 대해 알아보겠습니다.

 

💡 다형성을 이용하여 메시지 생성

• 우리는 클라이언트가 회원가입을 완료했을 경우 쿠폰 알림과 회원가입 축하에 대한 알림 2가지를 보내야합니다. 이를 다형성을 이용하여 만들겠습니다.
• EventHandler 인터페이스 상단에 있는 @JsonTypeInfo 어노테이션과 @JsonSubTypes 어노테이션은 스프링이 직렬화/역직렬화 하는 과정에서 구체 클래스를 알지 못하므로 오류를 내뱉게 되는데 이를 해결해줄 어노테이션입니다. 구체적인 사용방법은 구글링을 통해 자세히 알 수 있습니다.

@JsonTypeInfo(
        use = JsonTypeInfo.Id.NAME,
        property = "type",
        visible = true
)
@JsonSubTypes({
        @JsonSubTypes.Type(value = NotificationEventHandler.class, name = "notification"),
        @JsonSubTypes.Type(value = MemberCouponEventHandler.class, name = "memberCoupon"),
})
public interface EventHandler {

    // 문맥에 따라 변하는 부분을 구체 클래스에서 구체화
    NotificationResponse send(NotificationMessage message, NotificationService notificationService);
}

@Component
public class NotificationEventHandler implements EventHandler {

    @Override
    public NotificationResponse send(NotificationMessage message, NotificationService notificationService) {
        Notification notification = notificationService.save(message);
        return new NotificationResponse(notification.getMember().getPhone(), notification.getContents());
    }
}

@Component
public class MemberCouponEventHandler implements EventHandler {

    @Override
    public NotificationResponse send(NotificationMessage message, NotificationService notificationService) {
        Notification notification = notificationService.save(message);
        return new NotificationResponse(notification.getMember().getPhone(), notification.getContents());
    }
}

 

💡 메시지를 보낼 DTO 클래스

• 해당 DTO 클래스는 RabbitMQ에게 전달할 메시지입니다.

@Getter
@NoArgsConstructor
@AllArgsConstructor
public class NotificationMessage<T> {

    private EventHandler eventHandler;
    private NotificationType notificationType;
    private T data;

    public NotificationResponse notificationSend(NotificationService notificationService) {
        return eventHandler.send(this, notificationService);
    }
}

 

💡 이벤트 발행

• 다시 돌아가서 클라이언트가 회원가입을 완료하면 이벤트를 발행합니다.

@Service
@Transactional
@RequiredArgsConstructor
public class MemberService {

    private final MemberRepository memberRepository;
    private final ApplicationEventPublisher applicationEventPublisher;

    public void create(MemberCreateDTO request) {
        Member member = Member.builder()
                .name(request.getName())
                .phone(request.getPhone())
                .build();

        memberRepository.save(member); // 회원가입 완료
        applicationEventPublisher.publishEvent(new NotificationMessage<>(new NotificationEventHandler(), NotificationType.JOIN, member));
        applicationEventPublisher.publishEvent(new NotificationMessage<>(new MemberCouponEventHandler(), NotificationType.COUPON, member));
    }
}

 

💡 RabbitMQ에게 메시지 보내기

• @TransactionalEventListener 어노테이션을 사용하여 트랜잭션이 커밋된 이후 작동되도록 합니다.
• 클라이언트가 정상적으로 회원가입이 완료되었다는 것을 캐치하게 된다면 RabbitMQ에게 메시지를 보냅니다.

@Component
@RequiredArgsConstructor
public class RabbitMQProducer {

    private final RabbitTemplate rabbitTemplate;
    private static final String EXCHANGE_NAME = "fanoutExchange";

    @TransactionalEventListener
    public void eventHandler(NotificationMessage notificationMessage) {
        sendMessage(notificationMessage);
    }

    private void sendMessage(NotificationMessage response) {
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, "", response);
    }
}

 

💡 RabbitMQ에게서 메시지 받기

• @RabbitListener 어노테이션을 사용하여 특정 큐를 경청하고 있습니다. 클라이언트가 회원가입이 완료되었다는 메시지가 오면 데이터 베이스 알림을 저장하고 특정 클라이언트에게 알림을 발송합니다.

@Component
@RequiredArgsConstructor
public class RabbitMQConsumer {

    private final RabbitTemplate rabbitTemplate;
    private static final String EXCHANGE_NAME = "amq.topic";
    private final ObjectMapper objectMapper;
    private final NotificationService notificationService;

    @RabbitListener(queues = "server-queue")
    public void consume(NotificationMessage notification) {
        System.out.println(notification);
        NotificationResponse response = notification.notificationSend(notificationService);
        sendMessage(response);
    }

    private void sendMessage(NotificationResponse response) {
        try {
            String json = objectMapper.writeValueAsString(response);
            rabbitTemplate.convertAndSend(EXCHANGE_NAME, response.getRoutingKey(), json);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }
}

 

💡 전달 받은 메시지를 데이터 베이스에 보관하기

• 아래 NotificationService는 단순합니다. 전달받은 메시지를 데이터 베이스에 저장하고 Notification 객체를 다시 반환하고 있습니다. 반환된 Notification은 RabbitMQConsumer 클래스에서 다시 전달받아 특정 클라이언트에게 메시지를 발송하게 됩니다.

@Service
@Transactional
@RequiredArgsConstructor
public class NotificationService {

    private final ObjectMapper objectMapper;
    private final NotificationRepository notificationRepository;

    public Notification save(NotificationMessage message) {
        Member member = objectMapper.convertValue(message.getData(), Member.class);
        String contents = messageContents(message.getNotificationType(), member.getName());

        Notification notification = Notification.builder()
                                .notificationType(message.getNotificationType())
                                .contents(contents)
                                .member(member)
                                .build();
        notificationRepository.save(notification);
        return notification;
    }

    private String messageContents(NotificationType notificationType, String name) {
        switch (notificationType) {
            case JOIN:
                return name + "님 가입을 축하드립니다.";
            case COUPON:
                return name + "님 쿠폰을 보내드립니다.";
            default:
                throw new IllegalArgumentException("처히할 수 없는 매개변수 입니다.");
        }
    }
}

 

정리

---

• 이로써 RabbitMQ를 사용하여 특정 클라이언트에게 알림을 보내는 예제가 끝이 났습니다. 여기서 회원가입 실패시 클라이언트의 정보는 롤백되고 알림은 발송되지 않습니다. 하지만 클라이언트는 정상적으로 회원가입이 이루어졌고 알림을 저장하는 상황에서 예외가 발생한다면 어떻게 될까요? 다음편에는 알림을 저장하는 상황에 예외가 발생했을 경우 어떻게 처리할 수 있는가에 대해 살펴보겠습니다.

 

RabbitMQ를 사용하여 알림 발송하기 2편

Spring Boot - RabbitMQ를 사용하여 알림 발송 2편