Redis로 초간단 Pub / Sub 구축하기! Redis + Spring Boot

일전에 Redis를 Spring Boot와 연동하여 캐시로 사용하는 방법에 대해 간략하게 알아보았습니다.

레디스(Redis)? 함 써봅시다. Spring Boot + Redis 연동하기!

 

📌 Pub/Sub은 왜 구축해보려 하는지?

몇 일 전 회사에서 이슈가 하나 있었습니다.

분당 30만건 정도의 트래픽이 발생하는 서비스 모듈에서 타 서비스 모듈로의 데이터 전달 시 지연이 발생하며 데이터의 실시간성이 훼손되는 문제였었습니다.

최대 10분정도까지 데이터 차이가 발생한다고 하더군요..ㄷㄷ

이유를 보니 기존에 데이터를 발생하고 저장하는 모듈이 RDB(Oracle)에 저장을 해주면 해당 RDB의 Table에서 데이터를 긁어가는 Agent가 있고 그 Agent가 타 서비스 모듈로 데이터를 문자열로 전달해주는 기능이었습니다.

 

아주 간단한 기능이지만 분당 30만건의 이벤트를 처리하려다보니 메모리 이슈가 좀 발생했었나봅니다.

거기에 정말 치명적인 이슈는 분당 RDB에서 통계용 쿼리가 돌아가고 있다는 것이었지요..

그 얘기는 RDB 서버가 매우매우 비지비지한 상태라는거죠. 그래서 30만건의 분량을 그저 읽어가기만 하면 되는 Agent 입자에선 데이터를 달라고해도 감감무소식이니 마냥 기다릴 수 밖에요..

언젠간 주겠지..

서비스 발생 모듈에서 원본데이터는 가지고 있어야 하며, 그 원본데이터는 타 모듈에서도 가져가야 하는 상황에 놓이다보니 새로운 무언가 데이터를 연계할 수 있는 것이 있어야겠다는 생각을 하게 되었고, 그리하야 몇 일 전 알아보았던 Redis의 Pub/Sub모델이 생각났습니다.

 

사실 여기서 '아니 그냥 MQ로 하면 되지 왜 Redis를?'라고 할 수 있을 것 같습니다.

우선 무엇보다 현재 고객사에서 구동되고 있는 서비스에 영향을 주지 않으면서 가장 간편하게 확장 가능하도록 변경을 해야하는 상황이었기 때문에 Kafka나 MQ를 도입하기에는 한계가 있다고 생각이 들었습니다.

(이래놓고 추후에 바뀔 수도 있지요)

 

추후에 KSQLDB 관련해서도 올리겠지만, 우선 이번엔 가벼운 확장을 위해 Redis를 적용하기로 했습니다.

프로토 타입 개발을 해보도록 하죠.

 

구현할 구조는 이런 식입니다. 다만 하나의 Publisher와 하나의 Subscriber가 있는 1:1 구조로 개발을 할 것 입니다.

 

📌 프로젝트 구현

참고로 해당 프로젝트는 프로토 타입이기 때문에 Publisher와 Subscriber를 한 곳에 모아놓았습니다.

추후에는 별도 서비스로 따로 개발하여 사이트에 적용 될 예정입니다.

 

의존성은 다음처럼 추가해주시면 됩니다.

...(생략)...
dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-data-redis'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    compileOnly 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
}
...(생략)...

Web은 굳이 추가 안해도 되는데..Controller로도 테스트를 해보고자 추가시켜 놓았습니다.

 

우선 Redis 설정 정보를 가지고 Configuration을 만들어보겠습니다.

@Configuration
public class EventConfig {
	// 사용할 Channel Topic을 설정
	@Bean
	public ChannelTopic eventTopic() {
		return new ChannelTopic("dummyTopic");
	}
    
	// RedisTemplate 설정 정보 등록
	@Bean
	public RedisConnectionFactory eventRedsConnectionFactory() {
		RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration("127.0.0.1", 6379);
		return new LettuceConnectionFactory(configuration);
	}

	@Bean
	public RedisTemplate<String, String> eventRedisTemplate() {
		RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
		redisTemplate.setConnectionFactory(eventRedsConnectionFactory());
		redisTemplate.setKeySerializer(new StringRedisSerializer());
		redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<EventMessage>(EventMessage.class));
		return redisTemplate;
	}

	// Listener Container 설정 정보 등록
	@Bean
	public RedisMessageListenerContainer redisMessageListenerContainer() {
		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
		container.setConnectionFactory(eventRedsConnectionFactory());
		container.addMessageListener(eventListener(), eventTopic());
		return container;
	}

	@Bean
	public MessageListener eventListener() {
		return new EventListener(eventRedisTemplate());
	}

}

 

다음은 메시지의 발행 주체인 Publisher를 구현해줍니다.

// Message 발행 주체인 Publisher 등록
// sendMessage의 convertAndSend를 통해 Topic에 메시지를 전송하게 됩니다.
@Slf4j
@Component
@RequiredArgsConstructor
public class EventPublisher {
	private final RedisTemplate<String, String> eventRedisTemplate;
	private final ChannelTopic eventTopic;

	public void sendMessage(EventMessage eventMessage) {
		eventRedisTemplate.convertAndSend(eventTopic.getTopic(), eventMessage);
	}
}

 

다음은 메시지 소비의 주체인 Listener를 구현해줍니다.

@Slf4j
public class EventListener implements MessageListener {
	private RedisTemplate<String, String> eventRedisTemplate;
	private RedisSerializer<EventMessage> valueSerializer;

	public EventListener(RedisTemplate<String, String> eventRedisTemplate) {
		this.eventRedisTemplate = eventRedisTemplate;
		this.valueSerializer = (RedisSerializer<EventMessage>) eventRedisTemplate.getValueSerializer();
	}

	@Override
	public void onMessage(Message message, byte[] pattern) {
		EventMessage eventMessage = valueSerializer.deserialize(message.getBody());
		log.warn("Sub Channel : {}", new String(message.getChannel()));
		log.warn("Sub Msg : {}", eventMessage.toString());
	}
}

valueSerializer에 EventMessage로 변환하도록 정의하였습니다. 이를 통해 손쉽게 Topic 내 데이터를 EventMessage로 만들어 사용할 수 있습니다!

Log를 통해 메시지를 전달받은 Topic Channel 정보와 Message를 찍어줍니다.

EventMessage Entity는 다음과 같습니다.

@Getter
@ToString
public class EventMessage {
	private Long timestamp;
	private String message;

	public EventMessage(String message) {
		this.timestamp = System.currentTimeMillis();
		this.message = message;
	}

	@JsonCreator
	public EventMessage(@JsonProperty("timestamp") Long timestamp,
						@JsonProperty("message") String message) {
		this.timestamp = System.currentTimeMillis();
		this.message = message;
	}
}

 

준비가 끝났으니 이제 테스트를 해보도록 하겠습니다.

TestClass를 다음과 같이 구현해줍니다.

@SpringBootTest
class RedisPubSubApplicationTests {
	@Autowired
	private EventPublisher eventPublisher;

	@Test
	public void testPubSub() throws InterruptedException {
		eventPublisher.sendMessage(new EventMessage("Redis Pub/Sub Test Message"));
		TimeUnit.SECONDS.sleep(3);
	}

}

 

테스트가 통과되고 다음처럼 console에 log가 찍히게 됩니다.

2023-12-29T15:22:05.370+09:00  WARN 43528 --- [enerContainer-1] c.e.r.domain.event.EventListener         : Sub Channel : dummyTopic
2023-12-29T15:22:05.371+09:00  WARN 43528 --- [enerContainer-1] c.e.r.domain.event.EventListener         : Sub Msg : EventMessage(timestamp=1703830925370, message=Redis Pub/Sub Test Message)

 

📌 마치며

아주 간단하게 구현해본 Redis Pub/Sub 아키텍처 프로젝트입니다. 이를 통해 데이터 파이프라이닝을 아주 손쉽게 구현할 수 있습니다.

간단하게 인메모리 DB의 빠른 성능을 체감하고 싶다면 사용해보기에 좋을 것 같습니다.