spring boot redis pub sub

Spring Boot의 redis pub sub을 알아보자. 저번에는 레디스 캐싱에 대해 알아 봤는데..약간 부족한 면이 있다 나중에 좀더 살펴 봐야 겠다. pub sub 시스템은 채널에 구독 신청을 한 모든 sub에게 메시지를 전달한다. 한마디로 구독한 채널에 대해 메시지를 전달하는 시스템을 말하는거다. Spring boot를 사용하면 좀더 쉽게 셋팅하고 구현할 수 있다. 일단 레디스가 설치 되어 있어야 한다. 레디스 설치는 인터넷에서 참고하기 바란다. 바로 소스를 보자 일단 maven으로 하기 때문에 pom xml 파일을 보자.

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.7.3</version>
</dependency>

<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>

일단 boot의 redis를 사용하기 위해 starter-data-redis 와 json형태로 전달하기 위해 jackson을 넣어 두었다.

public class JsonRedisTemplate<V> extends RedisTemplate<String, V> {

  public JsonRedisTemplate(RedisConnectionFactory connectionFactory, ObjectMapper objectMapper, Class valueType) {
    RedisSerializer<String> stringSerializer = new StringRedisSerializer();
    super.setKeySerializer(stringSerializer);
    super.setHashKeySerializer(stringSerializer);
    super.setHashValueSerializer(stringSerializer);
    Jackson2JsonRedisSerializer jsonRedisSerializer = new Jackson2JsonRedisSerializer<>(valueType);
    jsonRedisSerializer.setObjectMapper(objectMapper);
    super.setValueSerializer(jsonRedisSerializer);
    super.setConnectionFactory(connectionFactory);
    super.afterPropertiesSet();
  }
}

Json으로 전달을 하기 위해서 JsonRedisTemplate만들어 setValueSerializer에 json Serializer를 셋팅 하였다. 이번엔 redis에 필요한 빈 설정들을 보자

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

  @Bean
  public TopicMessageListener topicMessageListener() {
    return new TopicMessageListener();
  }

  @Bean
  MessageListenerAdapter listenerAdapter(MethodMessageListener methodMessageListener) {
    return new MessageListenerAdapter(methodMessageListener, "sendMessage");
  }

  @Bean
  MethodMessageListener methodMessageListener() {
    return new MethodMessageListener();
  }

  @Autowired
  private JsonRedisTemplate jsonRedisTemplate;

  @Bean
  public ObjectMapper objectMapper(){
    return new ObjectMapper();
  }

  @Bean
  public JsonRedisTemplate jsonRedisTemplate(RedisConnectionFactory connectionFactory, ObjectMapper objectMapper){
    return new JsonRedisTemplate<>(connectionFactory,objectMapper,Object.class);
  }

우리는 두가지 방법으로 설정 해보겠다. 실제 메소드를 이용해보는 방식이 하나 있고 다음은 토픽의 키 값을 이용해 보는 방식으로 해보겠다. 사실은 두개의 인터페이스는 같다. 첫번째에 있는 빈은 메시지 리스너를 등록하는 빈이다. 우리는 두가지의 방법으로 해본다고 했으니 두개를 등록 시켜놨다. 다음에 있는 TopicMessageListener 은 MessageListener 인터페이스를 구현한 구현체이다. 그 다음은 메소드 방식으로 설정 하기 위해 빈으로 등록했다. 어댑터의 첫번째 파라미터는 sub의 클래스이다. 두번째는 메소드 명으로 등록하면 된다. 그리고 아까 만든 JsonRedisTemplate을 빈으로 등록 했다. 메소드 방식의 구현체인 MethodMessageListener 소스이다.

@Slf4j
public class MethodMessageListener {

  public void sendMessage(String message) {
    log.info(message);
  }
}

그냥 간단하게 출력 정도만 한다. 다음으로는 토픽의 키 값으로 하는 방식이다.

@Slf4j
public class TopicMessageListener implements MessageListener {

  private RedisTemplate redisTemplate;

  @Autowired
  public void setRedisTemplate(RedisTemplate redisTemplate) {
    this.redisTemplate = redisTemplate;
  }

  @Override
  public void onMessage(Message message, byte[] bytes) {
    byte[] body = message.getBody();
    String str = (String) redisTemplate.getStringSerializer().deserialize(body);
    log.info("key : {}, message : {}", new String(bytes), str);
  }
}

이것또한 출력만 하는 형식으로 만들었다. 아까 본 MessageListenerAdapter 클래스도 MessageListener의 구현체로 만들어져 있다. 그럼 테스트를 해보자.

  @Autowired
  private JsonRedisTemplate jsonRedisTemplate;

  @Test
  public void pubsub() {
    SendMessage sendMessage = new SendMessage();
    sendMessage.setName("wonwoo");
    sendMessage.setEmail("wonwoo@test.com");
    jsonRedisTemplate.convertAndSend("sendMessage", sendMessage);
  }

이렇게 테스트를 해보면 다음과 같이 로그가 남겨질 것이다.

2016-04-23 20:32:25.050  INFO 1151 --- [    container-2] me.wonwoo.redis.MethodMessageListener    : {"name":"wonwoo","email":"wonwoo@test.com"}
2016-04-23 20:32:25.050  INFO 1151 --- [    container-3] me.wonwoo.redis.TopicMessageListener     : key : sendMessage, message : {"name":"wonwoo","email":"wonwoo@test.com"}

MethodMessageListener클래스에서 한번, TopicMessageListener 클래스에서 한번 로그가 출력 될 것이다. 이번엔 redis-cli을 실행 시켜 보자. 접속후에 다음과 같이 입력해보자

SUBSCRIBE sendMessage

그리고 테스트를 다시 돌려보면

"{\\"name\\":\\"wonwoo\\",\\"email\\":\\"wonwoo@test.com\\"}"

이런 형태의 스트링 Json이 출력 될 것이다. 이상으로 Spring의 redis pub sub를 알아 봤다! 간단하다! github에 올려 놔야겠당.