[SpringBoot] Redis Channel Subscribe with MessagePack

Spring Boot 환경에서 Redis로부터 MessagePack으로 압축되어 전송된 문자열을 복호화 하느라 며칠 삽질을 했었다. 최신 릴리즈 버전과 관련 자료도 없었기에 더 고생을 했는데, 나 같은 사람이 더 이상 발생하지 않길 빌며 공유해본다. 😅

 

Spring Boot에서 Redis를 사용하고 있고, 특정 Channel을 Subscribe 하고 있을 때, Publish 되어 전송되는 메시지가 MessagePack 방식으로 압축이 되어있는 경우 다음과 같이 압축을 해제하여 메시지 원문을 받을 수 있다.

* MessagePack 이란?

MessagePack: It's like JSON. but fast and small.

MessagePack is an efficient binary serialization format. It lets you exchange data among multiple languages like JSON. But it's faster and smaller. Small integers are encoded into a single byte, and typical short strings require only one extra byte in addition to the strings themselves.

먼저 MessagePack 관련 라이브러리를 다운받자.

    // Message Pack For Redis
    compile 'org.msgpack:msgpack-core:0.8.18'
    compile 'org.msgpack:jackson-dataformat-msgpack:0.8.18'

MessagePack으로 압축되어 Redis에서 특정 채널로 전송한 메시지를 복호화하기 위해서는 MessageListener의 구현체를 MessageListenerAdapter로 등록하면 안 된다. MessageListener 인터페이스에서는 메시지를 받아 처리하는 메소드를 다음과 같이 정의했기 때문이다.

package org.springframework.data.redis.connection;

import org.springframework.lang.Nullable;

/**
 * Listener of messages published in Redis.
 *
 * @author Costin Leau
 * @author Christoph Strobl
 */
public interface MessageListener {

	/**
	 * Callback for processing received objects through Redis.
	 *
	 * @param message message must not be {@literal null}.
	 * @param pattern pattern matching the channel (if specified) - can be {@literal null}.
	 */
	void onMessage(Message message, @Nullable byte[] pattern);
}

public interface Message extends Serializable {

	/**
	 * Returns the body (or the payload) of the message.
	 *
	 * @return message body. Never {@literal null}.
	 */
	byte[] getBody();

	/**
	 * Returns the channel associated with the message.
	 *
	 * @return message channel. Never {@literal null}.
	 */
	byte[] getChannel();
}

압축된 문자열의 메시지를 RedisMessageListenerContainer에서 Message 형태로 변환하고 이를 주입하기 때문이다.

이를 해결하는 방법은 Delegate 객체를 만들어 MessageListenerAdapter에 등록해주는 방법이 있다. 다음은 Spring Data Redis의 API 문서 내용 중 Delegate 객체와 관련된 내용이다.

 

밑줄 친 부분을 주로 보시면 됩니다. 

public class MessageListenerAdapter extends Object implements InitializingBean, MessageListener
Message listener adapter that delegates the handling of messages to target listener methods via reflection, with flexible message type conversion. Allows listener methods to operate on message content types, completely independent from the Redis API.
Make sure to call afterPropertiesSet() after setting all the parameters on the adapter.
Note that if the underlying "delegate" is implementing MessageListener, the adapter will delegate to it and allow an invalid method to be specified. However if it is not, the method becomes mandatory. This lenient behavior allows the adapter to be used uniformly across existing listeners and message POJOs.
Modeled as much as possible after the JMS MessageListenerAdapter in Spring Framework.
By default, the content of incoming Redis messages gets extracted before being passed into the target listener method, to let the target method operate on message content types such as String or byte array instead of the raw Message. Message type conversion is delegated to a Spring Data RedisSerializer. By default, the JdkSerializationRedisSerializer will be used. (If you do not want such automatic message conversion taking place, then be sure to set the Serializer to null.)

 

 

Find below some examples of method signatures compliant with this adapter class. This first example handles all Message types and gets passed the contents of each Message type as an argument.

 

 

 public interface MessageContentsDelegate {
        void handleMessage(String text);

        void handleMessage(byte[] bytes);

        void handleMessage(Person obj);
 }
 

 

 

In addition, the channel or pattern to which a message is sent can be passed in to the method as a second argument of type String:

 

 

 public interface MessageContentsDelegate {
        void handleMessage(String text, String channel);

        void handleMessage(byte[] bytes, String pattern);
 }

 

 

For further examples and discussion please do refer to the Spring Data reference documentation which describes this class (and its attendant configuration) in detail.Important:Due to the nature of messages, the default serializer used by the adapter isStringRedisSerializer. If the messages are of a different type, change them accordingly throughsetSerializer(RedisSerializer).

Author: Juergen Hoeller, Costin Leau, Greg Turnquist, Thomas Darimont, Christoph Strobl, Mark Paluch

그럼 이제 Deletegate Interface를 정의해보고, Serializer를 null로 설정하고 MessagePack으로 압축된 메시지를 복호화해보자.

public interface MessageDelegate {
    void handleMessage(byte[] message, String channel) throws IOException;
}

 

@Service
public class MessagePackMessageDelegate implements MessageDelegate {
    @Override
    public void handleMessage(byte[] message, String channel) throws IOException {
        ObjectMapper objectMapper = new ObjectMapper(new MessagePackFactory());
        List<Object> deserialized = objectMapper.readValue(message, new TypeReference<List<Object>>() {
        });
        System.out.println(deserialized);
    }
}

 

publish 되는 메시지의 구조가 유동적으로 변했기 때문에 List <Object> 형태로 값을 읽었다. 만약 메시지의 형식이 구조화되어있다면 그에 맞는 DTO 객체를 만들어 다음과 같이 값을 읽으면 된다.

    @Getter
    @Setter
    public static class MessageDTO {
        private Long messageId;
        private String category;
        private String title;
        private String sender;
        private String message
    }

 

@Service
public class MessagePackMessageDelegate implements MessageDelegate {
    @Override
    public void handleMessage(byte[] message, String channel) throws IOException {
        ObjectMapper objectMapper = new ObjectMapper(new MessagePackFactory());
        MessageDTO deserialized = objectMapper.readValue(message, MessageDTO.class);
        System.out.println(deserialized);
    }
}

위와 같이 MessagePackMessageDelegate 클래스를 만들고 이를 다음과 같이 MessageListenerAdapter에 등록하고, RedisMessageListenerContainer에 구독할 채널 정보를 가진 Topic 클래스(ChannelTopic 또는 PatternTopic)와 함께 등록하면 된다.

    @Bean
    MessageListenerAdapter messageListenerAdapter(MessagePackMessageDelegate messagePackMessageDelegate) {
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter();
        messageListenerAdapter.setSerializer(null);
        messageListenerAdapter.setDelegate(messagePackMessageDelegate);
        messageListenerAdapter.afterPropertiesSet();
        return messageListenerAdapter;
    }

    @Bean
    PatternTopic topic() {
        return new PatternTopic("topic_*");
    }

    @Bean
    RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory , MessageListenerAdapter messageListenerAdapter, PatternTopic topic) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        container.addMessageListener(messageListenerAdapter, topic);
        return container;
    }