Redis Pub/Sub

Spring Data Reids에서 Redis Pub/Sub을 구현

• pub/sub은 message queueing system과 다르게 모든 구독자에게 데이터가 전달된다.
• Redis pub/sub의 경우 데이터를 메모리에 저장하지 않고 channel을 통해 전달하고 그 즉시 폐기되는 형식이다.

 

Publishing

Message를 전달하는 주체

• 일반적인 Operations의 convertAndSend를 통해 특정 채널에 데이터를 전달할 수 있다.

RedisExampleEntity redisExampleEntity = new RedisExampleEntity(1L, 1L, "CALL");
String serializedEntity = objectMapper.writeValueAsString(redisExampleEntity);
redisTemplate.**convertAndSend**("example-channel-name", serializedEntity);

 

Subscribing

Message를 받는 주체

• 메세지를 받을 때, 특정 채널 이름 뿐만 아니라 패턴 매칭을 통해서도 받을 수 있다.
  • 새롭게 추가되는 채널에 대해 restart 없이 바로 반영이 가능
• subscribe를 수행하는 thread는 blocking 상태가 되고, 다른 쓰레드에서 channel에 대해 unsubscribe을 하는 경우에만 blocking 상태가 해제된다.
  • 이러한 subscribe 시의 쓰레드 관리를 framework에서 관리하기 위해 MessageListenerContainers를 사용하는 것이 좋다.

 

Message Listener Containers

Subscriber가 기본적으로 thread를 blocking 시키기 때문에 개발자 입장에서 신경써야 하는 부분이 많아진다.

• 이러한 Thread 할당/해제 및 메세지 subscribing/dispatch를 해주는 별도의 MessageListenerContainers를 사용

또한 여러 message listener가 존재한다고 해도, Message Listener Containers 내부에서 하나의 ThreadPool이나 connection을 사용하기 때문에 구독하는 채널의 개수가 많아진다고 해도 리소스상의 부담이 적어진다.

기본적으로 MessageListenerContainers는 SimpleAsyncTaskExecutor를 사용한다.

• SimpleAsyncTaskExecutor는 각 요청마다 새로운 쓰레드를 생성해 비동기로 해당 테스크를 처리
• 작업이 끝난 테스크는 바로 쓰레드를 해제
• 동시에 처리될 수 있는 테스크의 수에 제한이 없기 때문에 생성될 수 있는 쓰레드의 개수도 제한이 없다.

이러한 SimpleAsyncTaskExecutor의 특징은 실제 서비스에서 문제가 될 수 있기 때문에 적절한 TaskExecutor를 사용하는 것이 중요

@Bean
TaskExecutor redisPubsubTaskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(1);
    taskExecutor.setMaxPoolSize(1);

    return taskExecutor;
}

@Bean
RedisMessageListenerContainer redisMessageListenerContainer(
        RedisConnectionFactory connectionFactory,
        MessageListener messageListener,
        MessageListenerAdapter messageListenerAdapter,
        TaskExecutor redisPubsubTaskExecutor
) {
    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    **container.setTaskExecutor(redisPubsubTaskExecutor);**
		
		...

    return container;
}

 

Message Listener

MessageListenerContainers를 통해 메세지를 처리할 수 있는 방법은 대표적으로 2가지가 있다.

1. MessageListener interface를 구현해 onMessage를 overriding
2. 일반 POJO 객체 + MessageListenerAdapter를 사용

 

MessageListener interface 구현

MessageListener interface를 구현해 바로 MessageListener를 만들 수 있다.

class MessageListenerImpl implements MessageListener {
    ObjectMapper objectMapper = new ObjectMapper();
    
    @Override
    public void onMessage(Message message, byte[] pattern) {
        String threadName = Thread.currentThread().getName();
        System.out.println("message received. current Thread : " + threadName);

        try {
            RedisExampleEntity redisExampleEntity = objectMapper.readValue(message.toString(), RedisExampleEntity.class);
            System.out.println("redisExampleEntity.id = " + redisExampleEntity.id);
            System.out.println("redisExampleEntity.callId = " + redisExampleEntity.callId);
            System.out.println("redisExampleEntity.status = " + redisExampleEntity.status);

        } catch (JsonProcessingException e) {
            System.out.println("error : " + e.getMessage());
            throw new RuntimeException(e);
        }
    }
}

이렇게 만들어진 MessageListener를 MessageListenerContainer에 입력해 특정 채널의 message를 받도록 할 수 있다.

@Bean
RedisMessageListenerContainer redisMessageListenerContainer(
        RedisConnectionFactory connectionFactory,
        **MessageListener messageListener,**
        MessageListenerAdapter messageListenerAdapter,
        TaskExecutor redisPubsubTaskExecutor
) {
		...

    **container.addMessageListener(messageListener, new ChannelTopic("test1"));**

    return container;
}

 

POJO 사용

위의 MessageListenerImpl은 MessageListener interface를 구현한 클래스이다.

• 어떠한 디팬던시 없이 POJO를 통해서 메세지를 처리할 수 있다.

1. POJO interface 정의

• 이때 메세지를 처리할 메소드를 같이 정의해줘야 한다.

interface POJOMessageListener {
		// 기타 필요한 메서드
		...

		// 메세지 처리에 필요한 메서드
    void handleMessage(String message, byte[] pattern);
    void handleMessage(String message, String channel);
}

 

 

2. POJO interface를 구현한 클래스 정의

class POJOMessageListenerImpl implements POJOMessageListener {

    ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void handleMessage(String message, byte[] pattern) {
        handleMessage(message, new String(pattern));
    }

    @Override
    public void handleMessage(String message, String channel) {
        System.out.println("pojo message received. current channel : " + channel);

        try {
            RedisExampleEntity redisExampleEntity = objectMapper.readValue(message, RedisExampleEntity.class);
            System.out.println("redisExampleEntity.id = " + redisExampleEntity.id);
            System.out.println("redisExampleEntity.callId = " + redisExampleEntity.callId);
            System.out.println("redisExampleEntity.status = " + redisExampleEntity.status);

        } catch (JsonProcessingException e) {
            System.out.println("error : " + e.getMessage());
            throw new RuntimeException(e);
        }
    }
}

 

3. MessageListenerAdapter 구현

• POJO 클래스를 messageListener처럼 사용할 수 있도록 해주는 어댑터
• 이때 위에서 정의한 handleMessage와 같은 메소드를 메세지 처리 시 사용하도록 명시

@Bean
MessageListenerAdapter messageListenerAdapter() {
    return new MessageListenerAdapter(
            new POJOMessageListenerImpl(),
            "handleMessage" // 해당 메소드를 메시지 처리 시 사용하도록 정의
    );
}

 

4. MessageListenerContainer에 등록

@Bean
RedisMessageListenerContainer redisMessageListenerContainer(
        RedisConnectionFactory connectionFactory,
        MessageListener messageListener,
        MessageListenerAdapter messageListenerAdapter,
        TaskExecutor redisPubsubTaskExecutor
) {
		...

    container.addMessageListener(messageListener, new ChannelTopic("test1"));
		container.addMessageListener(messageListenerAdapter, new ChannelTopic("test2"));

    return container;
}

 

전체 코드

messageListener

class MessageListenerImpl implements MessageListener {
    ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void onMessage(Message message, byte[] pattern) {
				System.out.println("message received. current channel : " + new String(message.getChannel()) );

        try {
            RedisExampleEntity redisExampleEntity = objectMapper.readValue(message.toString(), RedisExampleEntity.class);
            System.out.println("redisExampleEntity.id = " + redisExampleEntity.id);
            System.out.println("redisExampleEntity.callId = " + redisExampleEntity.callId);
            System.out.println("redisExampleEntity.status = " + redisExampleEntity.status);

        } catch (JsonProcessingException e) {
            System.out.println("error : " + e.getMessage());
            throw new RuntimeException(e);
        }

    }
}

interface POJOMessageListener {
    void handleMessage(String message, byte[] pattern);

    void handleMessage(String message, String channel);

}

class POJOMessageListenerImpl implements POJOMessageListener {

    ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void handleMessage(String message, byte[] pattern) {
        handleMessage(message, new String(pattern));
    }

    @Override
    public void handleMessage(String message, String channel) {
        System.out.println("pojo message received. current channel : " + channel);

        try {
            RedisExampleEntity redisExampleEntity = objectMapper.readValue(message, RedisExampleEntity.class);
            System.out.println("redisExampleEntity.id = " + redisExampleEntity.id);
            System.out.println("redisExampleEntity.callId = " + redisExampleEntity.callId);
            System.out.println("redisExampleEntity.status = " + redisExampleEntity.status);

						System.out.println();
            System.out.println();

        } catch (JsonProcessingException e) {
            System.out.println("error : " + e.getMessage());
            throw new RuntimeException(e);
        }
    }
}

 

RedisConfig

@Configuration
public class RedisConfig {

    @Bean
    RedisConnectionFactory lettuceConnectionFactory() {
        return new LettuceConnectionFactory(
                new RedisStandaloneConfiguration("localhost", 6379)
        );
    }

    @Bean(name = "RedisStringTemplate")
    RedisTemplate<String, String> redisStringTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<String, String> template = new StringRedisTemplate();
        template.setConnectionFactory(connectionFactory);
        template.setEnableTransactionSupport(true);

        return template;
    }

    @Bean
    TaskExecutor redisPubsubTaskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(1);
        taskExecutor.setMaxPoolSize(1);

        return taskExecutor;
    }

    @Bean
    MessageListenerAdapter messageListenerAdapter() {
        return new MessageListenerAdapter(
                new POJOMessageListenerImpl(),
                "handleMessage"
        );
    }

    @Bean
    MessageListener messageListener() {
        return new MessageListenerImpl();
    }

    @Bean
    RedisMessageListenerContainer redisMessageListenerContainer(
            RedisConnectionFactory connectionFactory,
            MessageListener messageListener,
            MessageListenerAdapter messageListenerAdapter,
            TaskExecutor redisPubsubTaskExecutor
    ) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setTaskExecutor(redisPubsubTaskExecutor);

        container.addMessageListener(messageListener, new ChannelTopic("test1"));
        container.addMessageListener(messageListenerAdapter, new ChannelTopic("test2"));
        return container;
    }
}

 

Test

@SpringBootTest
public class RedisPubSubExampleTest {
    @Autowired
    @Qualifier("RedisStringTemplate")
    RedisTemplate<String, String> redisTemplate;

    @Test
    void basicPubsubTest() throws JsonProcessingException {
        ObjectMapper objectMapper = new ObjectMapper();


        for (int i = 0; i < 3; i++) {
            Thread.sleep(100);
            RedisExampleEntity redisExampleEntity = new RedisExampleEntity((long) (i + 1), (long) (i + 1), "CALL");
            String serializedEntity = objectMapper.writeValueAsString(redisExampleEntity);
            redisTemplate.convertAndSend("test1", serializedEntity);
            redisTemplate.convertAndSend("test2", serializedEntity);
        }
    }
}

 

 

결과

message received. current channel : test1
redisExampleEntity.id = 1
redisExampleEntity.callId = 1
redisExampleEntity.status = CALL
pojo message received. current channel : test2
redisExampleEntity.id = 1
redisExampleEntity.callId = 1
redisExampleEntity.status = CALL

message received. current channel : test1
redisExampleEntity.id = 2
redisExampleEntity.callId = 2
redisExampleEntity.status = CALL
pojo message received. current channel : test2
redisExampleEntity.id = 2
redisExampleEntity.callId = 2
redisExampleEntity.status = CALL

message received. current channel : test1
redisExampleEntity.id = 3
redisExampleEntity.callId = 3
redisExampleEntity.status = CALL
pojo message received. current channel : test2
redisExampleEntity.id = 3
redisExampleEntity.callId = 3
redisExampleEntity.status = CALL