Spring Boot 기반 Redis Pub/Sub 구현: 분산 환경에서 Pod 간 이벤트 처리

목적

오늘은 Spring Boot 기반 프로젝트에서 Redis Pub/Sub을 설정하는 코드를 구현하려고 한다.

 

환경

분산 처리 환경(API들이 Pod로 분산 처리됨)에서 Redis를 Session Storage로 사용한다.

 

목표

특정 Pod에서 이벤트가 발생하면, Redis Pub/Sub을 통해 모든 Pod에 이벤트를 전달하는 구조를 구현한다.

 

구현

채널 구독

Redis Config 파일에 RedisMessageListenerContainer 를 Bean 등록해서 아래와 같이 PatternTopic 으로 채널을 구독한다.

여기서 채널을 구독하지 않으면 메세지를 수신할 수 없으니 유의하자.

 

@Bean
public RedisMessageListenerContainer redisContainer(RedisConnectionFactory connectionFactory,
                                                    MessageListenerAdapter listenerAdapter) {
    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.addMessageListener(listenerAdapter, new PatternTopic("SOOBIN_CHANNEL"));
    return container;
}

 

 

참고로 단일 또는 다중 채널을 구독할 수 있다.

new PatternTopic("SOOBIN_CHANNEL*")  > SOOBIN_CHANNEL 으로 시작하는 채널을 모두 구독한다.

new PatternTopic("SOOBIN_CHANNEL") > SOOBIN_CHANNEL 과 일치하는 채널을 모두 구독한다.

 

메세지 발신

"SOOBIN_CHANNEL" 채널에 메세지를 발행한다.

 

private final RedisTemplate<String, Object> redisTemplate;

public void publishToChannel(Object message) {
    redisTemplate.convertAndSend("SOOBIN_CHANNEL", message);
}

 

 

메세지 수신

"SOOBIN_CHANNEL" 채널을 구독하고 메세지를 수신한다.

 

MessageListener 를 상속 받아서 메세지를 수신하는 RedisMessageSubscriber.onMessage 를 구현한다.

CANNEL_PREFIX 와 일치하는 채널에서 메세지가 수신되면 RedisMessageSubscriber.onMessage 에서 메세지를 받을 수 있다.

 

MessageListener의 주된 역활은 아래와 같다.
- 메세지 구독메세지 브로커에서 발행된 메세지를 구독한다.
- 메세지 처리수신된 메세지에 대한 로직을 수행한다.
- 비동기 처리메세지를 비동기적으로 처리하므로 수신자는 메세지가 들어올 때까지 대기 상태에 있지 않는다.

 

@Service
@RequiredArgsConstructor
public class RedisMessageSubscriber implements MessageListener {

    private final static String CHANNEL_PREFIX = "SOOBIN_CHANNEL";

    /**
     * 메세지가 발행되면 이벤트를 받는 역활
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {

        // 채널
        String channel = new String(message.getChannel());

        // 메세지 내용
        String messageContent = new String(message.getBody());

    }
}

 

 

MessageListener 를 상속받은 RedisMessageSubscriber 을 Redis Confing Class 에 MessageListenerAdpter 를 Bean 으로 등록해야한다.

 

@Bean
public MessageListenerAdapter messageListener(RedisMessageSubscriber subscriber) {
    return new MessageListenerAdapter(subscriber);
}