Redis In Java_메시지 큐

https://123okk2.tistory.com/344

Redis In Java_자료구조

 

윗글에서 이어서 포스팅을 진행하자.

 

남이 만든 프로그램을 인수인계 받는 것은 참 어려운 일이다.

구조를 하나하나 다 뜯어보고, 코드 내부도 전부 읽어봐야 하니까 말이다.

 

아주 힘든 일이지만 그래도 그만큼 얻는 건 많은 것 같다.

만약 이번에 해당 프로그램을 인수인계 받지 않았다면

Redis를 메시지 큐로도 활용할 수 있다는 사실을 알지 못했을 테니까.

*함께 설치되는 인프라 중에 RabbitMQ도 있는데 굳이 왜 Redis를 썼을까 의문이 들기도 하지만.

 

어쨌든 Redis는 단순 키-값 구조의 인메모리 데이터베이스를 지원할 뿐 아니라

메시지 큐로써의 기능도 함께 지원하고 있다.

 

자바에서의 사용법은 다음과 같다.

 

프로젝트는 위의 링크에 있는 이전 포스팅의 것을 그대로 가져왔다.

 

 

 

 

 

Redis Config

 

메시지 큐라면 당연히 송신자와 수신자가 필요하다.

빠른 공부를 위해 한 프로젝트에 동시에 작성하였다.

 

최근 버전의 Spring Boot는 Redis 사용시 굳이 RedisTemplate을 설정해야할 필요가 없다.

하지만 메시지 큐 사용을 위해서는 어느정도의 설정은 해줘야 하는 모양이다.

 

먼저 수신자이다.

수신한 메시지를 처리할 클래스를 먼저 정의해야한다.

MessageListener을 상속받아야 하고,

간단하게 메시지 내용을 출력하고 List에 저장하는 코드를 작성해놓았다.

package com.example.demo;

import java.util.ArrayList;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;

public class Listener implements MessageListener{
	
	private Logger logger = LoggerFactory.getLogger(Listener.class);
	private List<Message> messages = new ArrayList<>();

	@Override
	public void onMessage(Message message, byte[] pattern) {
		logger.info("Message Received Channel : \t{}", new String(message.getChannel())); //byte[]로 오기 때문에 스트링으로 형변환이 필요하다.
		logger.info("Message Received Body : \t\t{}", new String(message.getBody()));
		messages.add(message); //수신한 메시지 저장
	}
}

 

다음에는 설정 클래스를 작성한다.

다른 여느 메시지 큐와 마찬가지로 Redis도 토픽을 중심으로 메시지를 송수신한다.

그렇기에 어떤 Topic을 구독할 것인지에 대한 명세가 필요했다.

package com.example.demo;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

@Configuration
public class RedisListenerConfig {
	
	@Value("${redisTopic}") private String topic;
	
	@Bean
	public Listener listener() {
		return new Listener();
	}
	
	@Bean
	public MessageListenerAdapter listenerAdapter(Listener listener) {
		return new MessageListenerAdapter(listener);
	}
	
	@Bean
	public RedisMessageListenerContainer container(RedisConnectionFactory connetcFactory, MessageListenerAdapter listenerAdapter) {
		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
		container.setConnectionFactory(connetcFactory);
		container.addMessageListener(listenerAdapter, new PatternTopic(topic));
		
		return container;
	}
}

addMessageListener 함수의 두 번째 파라미터가 바로 Topic이다.

 

이제 해당 Topic으로 들어온 주제들은 위에서 작성한대로 로그로 출력이될 것이다.

 

수신자는 여기까지가 끝이다.

그러면 이제 송신자를 만들어보자.

 

 

송신자는 굳이 거창한 config를 할 필요는 없다.

 

직접 Redis에 메시지를 송신하는 송신자(Publisher)와,

해당 송신자를 Bean으로 만들어 사용할 수 있도록하는 Bean 클래스을 만들어놓으면 끝이난다.

package com.example.demo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.stereotype.Service;

public class Publisher {
	
	@Autowired private StringRedisTemplate redisTemplate;
	@Value("${redisTopic}") private String topic;
	
	public void publish(String message) {
		redisTemplate.convertAndSend(topic, message);
	}

}

package com.example.demo;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RedisPublisherConfig {
	@Bean
	public Publisher redisPublisher() {
		return new Publisher();
	}
}

 

이제 application.properties에 토픽을 입력한 후 잘 돌아가는지에 대한 테스트를 수행한다.

redisTopic="testTopic"

 

테스트 코드는 아래와 같다.

@Autowired Publisher publisher;
	
@Test
void redisMessageTest() throws InterruptedException {
	publisher.publish("test Message 1");
	Thread.sleep(50);
	publisher.publish("test Message 2");
	Thread.sleep(50);
	publisher.publish("test Message 3");
	Thread.sleep(50);
	publisher.publish("test Message 4");
	Thread.sleep(50);
	publisher.publish("test Message 5");
}

굳이 sleep을 넣은 이유는 메시지가 너무 빠르게 수신되기 때문이다.

위에서 작성한 Listener 클래스에서 메시지 수신 시 topic과 body를 출력하게 되는데,

이 때 topic이 출력되고 body가 출력되기 전에 다시 수신된 topic이 출력되어 일부러 넣어주었다.

 

어쨌든 해당 코드의 결과는 아래이 잘 동작하게 된다.