Spring Boot Pub/Sub

2024.05.13 - [Spring/대용량 트래픽] - Spring Session

Spring Session

 

이전글에서는 세션으로써의 Redis를 알아봤다.

이번에는 Redis의 Pub/Sub 패턴에 대해 알아보겠다.

 

• 각 서버간의 느슨한 결합(loose coupling)을 위해 사용한다.
• 여기서 Redis는 메세지 브로커 역할을 해서 Publish(발행자) - Subscribe(수신인) 과의 중개 메세지 중개 한다.
• 그러므로 Publish는 수신할 대상을 몰라도된다.(Redis에만 전달하면됨!)
• Redis의 메세지 브로커 역할은 in-memory 기반 처리가 가능하고, 실시간처리상황에 적합하다.
• 가용성, 대량의 메세지와 같은 상황은 Kafka와 같이 다른시스템을 사용하는 대안이있다.
• Sub서버가 중간에 다운되었다가 다시 연결되어도 Redis는 메세지 정보를 저장하는 것이 아니므로 기존 전달 메세지는 Sub서버가 받을수 없다.

Pub/Sub 사용사례

1. 이벤트 전달(ex. 상품구매, 사용자탈퇴 ...etc)
2. 알림시스템
3. 라이브채팅 서비스

주요 명령어

• SUBSCRIBE [KEY]: 채널등록, 해당채널에 발행된 메세지 대기 및 수신
  • 해시 테이블 쓰기 때문에 O(1)로 동작
• PUBLISH [KEY] [VALUE]: 해당채널에 메세지 전달
• PUBSUB channels : 시스템내에서 등록된 채널들 조회
• PUBSUB numsub [KEY] : KEY에 해당하는 구독자 몇명?
• PSUBSCRIBE [KEY:*] : 패턴기반 채널등록 'KEY:' 뒤에 어떤것이 오던 공통으로 구독(SUB 입장임)
  • 리스트를 쓰기 때문에 O(n)으로 동작, 특별한 이유 없으면 SUBSCRIBE 그냥 쓰는게 성능상 이득

실습

PUB

SUB

SUB

여러개의 SUB에 잘 적용되는 걸 확인할수 있다.

 

PSUBSCRIBE를 실험보자

PUB

SUB

PUB에서 "users:" 까지만 일치하면 SUB에서 메세지를 받는것을 확인할수 있다.

구독자수 확인

---

Spring Data Redis를 활용한 PUB/SUB

 

SUBSCRIBE 코드

 

1. MessageListener 인터페이스의 onMessage()를 오버라이드로 서비스를 구현해야한다.
2. MessageListenerAdapter 인터페이스로 어떤 채널에 연결할지 구현한다. 위의 서비스를 인자로 받는다.
3. RedisMessageListenerContainer 를 만들어서  채널에 대한 listenerAdapter를 연결한다.

PUBLISH 코드

• RedisTemplate의 convertAndSend(채널, 메세지)를 호출하면 이벤트를 발행할 수 있다.

본격적으로 하기 위해 스프링시작 문서에서 프로젝트를 다운받아준다.

당연하게도 Spring Data Redis를 의존성 주입해줘야한다.

이제 서비스단부터 만들어본다.

MessageListenService.java

@Slf4j
@Service
public class MessageListenService implements MessageListener {
    @Override
    public void onMessage(Message message, byte[] pattern) {
        log.info("Received {} channel: {}", new String(message.getChannel()), new String(message.getBody()));
    }
}

 

간단하게 로그정보를 볼수 있게 코드를 짰다.

 

인자로 받는 Message는 무엇인가 직접들어가봤는데, 메세지 내용과, 채널명을 바이트 형태로 리턴하는 인터페이스이다.

Message.java

따라서 String으로 변환을시켜서 로그정보를 띄워준다.

 

다음으로는 설정파일이다.

RedisConfig.java

@Configuration
public class RedisConfig {
    @Bean
    MessageListenerAdapter messageListenerAdapter(){
        return new MessageListenerAdapter(new MessageListenService());
    }
    @Bean
    RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory,
                                                                MessageListenerAdapter listener){
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listener, ChannelTopic.of("users:unregister"));
        return container;
    }
}

MessageAdapter에 위에 만든 MessageService를 인자로 넣어주고, 어댑터 객체를 생성한다. 그리고 생성된 객체를 다시 RedisMessageListenerContainer에 인자로 넣어주어 연결된 Redis에 cli로 실습한다고 만들었던 users:unregister채널을 설정해주고 구독(Subscribe)연결한다.

 

이제 실행시키고, 잘 구독되었는지 Redis cli로 직접 실험해본다.

PUBLISH

SUBSCRIBE

잘 연동된것을 확인할수 있다.

 

---

PUBLISH JAVA 구현

 

Spring Data Redis를 RedisTemplate을 활용해서 API를 연동하겠다.

PublishController.java

@RestController
@RequiredArgsConstructor
public class PublishController {

    private final RedisTemplate<String , String> redisTemplate;

    @PostMapping("/events/users/deregister")
    void publishUserDeRegisterEvent(){
        redisTemplate.convertAndSend("users:unregister", "500");
    }
}

Postman으로 요청을 보내니 잘응답하는 것도 확인할수 있다.

 

이로써 Pub/Sub 역할의 Redis도 알아봤다. 

 

2024.05.21 - [Spring/대용량 트래픽] - Prometheus, Grafana 오픈소스로 Redis 모니터링하기

Prometheus, Grafana 오픈소스로 Redis 모니터링하기