[Redis] Pub/Sub (with Spring)

기본 예제

설명

• spring boot 프로젝트로 Embedded Redis를 띄우고 Pub/Sub 기능을 테스트하는 예제

의존성 추가

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>it.ozimov</groupId>
    <artifactId>embedded-redis</artifactId>
    <version>0.7.2</version>
</dependency>

application.properties

spring.profiles.active=local
spring.redis.host=localhost
spring.redis.port=6379

EmbeddedRedisConfig

@Profile("local")
@Configuration
public class EmbeddedRedisConfig {
    private RedisServer redisServer;

    @Value("${spring.redis.port}")
    private int port;

    @PostConstruct
    public void startRedis() {
        redisServer = new RedisServer(port);
        redisServer.start();
    }

    @PreDestroy
    public void stopRedis() {
        if (redisServer != null) {
            redisServer.stop();
        }
    }
}

ChatMessage

@Data
public class ChatMessage {
    private String senderId;
    private String message;
}

RedisConfig

@Configuration
public class RedisConfig {
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }

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

RedisPubSubController

@Slf4j
@AllArgsConstructor
@RestController
public class RedisPubSubController implements MessageListener {
    private final RedisTemplate<String, ChatMessage> redisTemplate;
    private final RedisMessageListenerContainer redisMessageListenerContainer;

    @PostMapping("/api/rooms/{roomId}")
    public void createChatRoom(@PathVariable String roomId) {
        redisMessageListenerContainer.addMessageListener(this, new ChannelTopic(roomId));
    }

    @DeleteMapping("/api/rooms/{roomId}")
    public void deleteChatRoom(@PathVariable String roomId) {
        redisMessageListenerContainer.removeMessageListener(this, new ChannelTopic(roomId));
    }

    @PostMapping("/api/rooms/{roomId}/chat")
    public void sendChatMessage(@PathVariable String roomId, @RequestBody ChatMessage chatMessage) {
        redisTemplate.convertAndSend(roomId, chatMessage);
    }

    @Override
    public void onMessage(Message message, byte[] bytes) {
        String roomId = new String(message.getChannel());
        ChatMessage chatMessage = (ChatMessage) redisTemplate.getValueSerializer().deserialize(message.getBody());

        log.info("[onMessage] roomId: {}, message: {}", roomId, chatMessage);
    }
}

채널 목록 확인

PUBSUB CHANNELS *

참고

• https://daddyprogrammer.org/post/4731/spring-websocket-chatting-server-redis-pub-sub/
• https://godd.tistory.com/38

단일 topic으로 관리하는 예제

설명

• 위의 기본 예제에서는 topic을 동적으로 추가/삭제하여 처리하도록 되어있음
• 채팅방이 여러개 생성될 경우 topic이 여러개 생성되고, 여러 topic의 메시지 수신은 onMessage 메소드 한군데서 처리하고 있음
• 각 room마다 topic을 부여하여 관리하는게 의미없으므로 하나의 topic을 두고, onMessage에서 각 room으로 분기 처리하는 방향으로 개선한 예제

RedisMessageListener

public interface RedisMessageListener extends MessageListener {
    String topic();
}

RedisConfig

@Configuration
public class RedisConfig {
    @Bean
    public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(connectionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        return redisTemplate;
    }

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(
        RedisConnectionFactory connectionFactory,
        List<RedisMessageListener> redisMessageListeners
    ) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        redisMessageListeners.forEach(listener -> container.addMessageListener(listener, new ChannelTopic(listener.topic())));
        return container;
    }
}

ChatController

@Slf4j
@AllArgsConstructor
@RestController
public class ChatController implements RedisMessageListener {
    private final RedisTemplate<String, String> redisTemplate;
    private final ObjectMapper objectMapper;

    @Override
    public String topic() {
        return "chat";
    }

    @SneakyThrows
    @PostMapping("/api/chat")
    public void publishMessage(@RequestBody ChatMessage chatMessage) {
        redisTemplate.convertAndSend(topic(), objectMapper.writeValueAsString(chatMessage));
    }

    @SneakyThrows
    @Override
    public void onMessage(Message message, byte[] bytes) {
        String topic = new String(message.getChannel());
        ChatMessage chatMessage = objectMapper.readValue(message.getBody(), ChatMessage.class);

        log.info("[onMessage] topic: {}, message: {}", topic, chatMessage);
    }

    @Data
    public static class ChatMessage {
        private String roomId;
        private String senderId;
        private String message;
    }
}

메시지 생성

curl --location --request POST 'http://localhost:8080/api/chat' \
--header 'Content-Type: application/json' \
--data-raw '{
    "roomId": "1234",
    "senderId": "john",
    "message": "Hello World"
}'

메시지 수신

[onMessage] topic: chat, message: ChatController.ChatMessage(roomId=1234, senderId=john, message=Hello World)