application.properties

spring.redis.host=focalhost-cluster
spring.redis.port=6379
spring.redis.topic.notice=notice

redisconfig

pub/sub으로 보낼 때는 redisTemplate을 사용하고

hash value를 넣고 뺄 때는 redisTemplateHash 넣고 빼는게 잘 되는 것 같다.


import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.jsontype.impl.LaissezFaireSubTypeValidator;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import io.lettuce.core.ReadFrom;
import io.lettuce.core.cluster.ClusterClientOptions;
import io.lettuce.core.cluster.ClusterTopologyRefreshOptions;
import lombok.RequiredArgsConstructor;
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.RedisClusterConfiguration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.Collections;
import java.util.List;

@RequiredArgsConstructor
@Configuration
public class RedisConfig {


    @Value("${spring.redis.host}")
    private String host;

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

    @Value("${spring.redis.topic.notice}")
    private String noticeTopic;


    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        List<String> nodes = Collections.singletonList(String.format("%s:%s", host, port));
        RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration(nodes);

        ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder().closeStaleConnections(true)
                .enableAllAdaptiveRefreshTriggers().build();

        ClusterClientOptions clusterClientOptions = ClusterClientOptions.builder().autoReconnect(true)
                .topologyRefreshOptions(topologyRefreshOptions).validateClusterNodeMembership(false)
                .build();

        LettuceClientConfiguration lettuceClientConfiguration = LettuceClientConfiguration.builder()
                .readFrom(ReadFrom.REPLICA_PREFERRED).clientOptions(clusterClientOptions).build();

        LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(clusterConfiguration, lettuceClientConfiguration);
        lettuceConnectionFactory.afterPropertiesSet();

        return lettuceConnectionFactory;
    }

    @Bean("redisTemplateHash")
    public RedisTemplate<?, ?> redisTemplateHash() {
        // hash 저장시 사용할 serializer, redisTemplate 사용하다가 안도면 이걸로 해보자.
        // 현재는 redis에 eventUtil 상태 저장을 위해서 사용 중이다.
        RedisTemplate<?, ?> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory());

        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance,ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.WRAPPER_ARRAY);
        objectMapper.registerModule(new JavaTimeModule());
        GenericJackson2JsonRedisSerializer genericJackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer(objectMapper);

        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(genericJackson2JsonRedisSerializer);
        redisTemplate.setHashKeySerializer(genericJackson2JsonRedisSerializer);
        redisTemplate.setHashValueSerializer(genericJackson2JsonRedisSerializer);
        return redisTemplate;
    }

    @Bean
    public RedisTemplate<?, ?> redisTemplate() {
        RedisTemplate<?, ?> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory());

        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);

        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        return redisTemplate;
    }


    @Bean
    public RedisMessageListenerContainer redisMessageListener(RedisConnectionFactory connectionFactory
            , MessageListenerAdapter noticeListenerAdapter) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(noticeListenerAdapter, new ChannelTopic(noticeTopic));

        return container;
    }



    @Bean
    public MessageListenerAdapter noticeListenerAdapter(NoticeSubscriber subscriber) {
        return new MessageListenerAdapter(subscriber, "receive");
    }


}

추상클래스

import com.fm.webinar.util.ByteUtil;
import lombok.Builder;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import lombok.experimental.SuperBuilder;

import java.sql.Timestamp;

@Getter
@NoArgsConstructor
@SuperBuilder(toBuilder = true)
public abstract class BaseMessage {

    @Builder.Default
    private String id = ByteUtil.newId();

    @NonNull
    private String sessionId;
    @NonNull
    private TopicType topicType;

    private Object extraData;

    @Builder.Default
    private Timestamp createTime = new Timestamp(System.currentTimeMillis());

}

구현 클래스

import com.fm.webinar.domain.message.base.BaseMessage;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import lombok.extern.jackson.Jacksonized;

import java.io.Serializable;

@Getter
@Jacksonized
@NoArgsConstructor
@SuperBuilder(toBuilder = true)
public class NoticeMessage extends BaseMessage implements Serializable {

    private static final long serialVersionUID = 1012808619350923621L;

    private String message;
}

pub

import com.fm.webinar.domain.message.notice.NoticeMessage;
import com.fm.webinar.domain.message.qna.QnaMessage;
import com.fm.webinar.util.ByteUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.Query;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
@RequiredArgsConstructor
@Service
public class NoticeService {

    @Value("${spring.redis.topic.notice}")
    private String topic;
    private final RedisTemplate redisTemplate;

    private final String NOTICE_INDEX_PREFIX = "notice_";

    public void send(NoticeMessage noticeMessage) {
        redisTemplate.convertAndSend(topic, noticeMessage);
    }

}

sub

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fm.webinar.domain.message.notice.NoticeMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;

@Slf4j
@RequiredArgsConstructor
@Service
public class NoticeSubscriber {
    private final ObjectMapper objectMapper;
    private final SimpMessagingTemplate template;

    public void receive(String noticeMessage) throws JsonProcessingException {
        log.info("@@@@@@@@@@ notice 구독함 @@@@@@@@@ {}", noticeMessage);
        NoticeMessage message = objectMapper.readValue(noticeMessage, NoticeMessage.class);
        this.template.convertAndSend("/sub/notice/" + message.getSessionId(), message);
    }
}

hash

import com.fm.webinar.domain.message.quiz.template.QuizTemplateMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.text.MessageFormat;


@Slf4j
@RequiredArgsConstructor
@Service
@Component
public class QuizService {

    private final RedisTemplate redisTemplate;

    private final RedisTemplate redisTemplateHash;

    private final String INDEX_TEMPLATE_PREFIX = "quiz_";

    @PostConstruct
    public void init() {
        hashOpsQuizRate = redisTemplate.opsForHash();
        hashOpsQuizTemplate = redisTemplateHash.opsForHash();
    }

    private HashOperations<String, String, QuizTemplateMessage> hashOpsQuizTemplate;


    public void save(QuizTemplateMessage message) {
        hashOpsQuizTemplate.put(MessageFormat.format("{0}{1}", INDEX_TEMPLATE_PREFIX, message.getSessionId())
                , message.getId(), message);
    }

    public QuizTemplateMessage getTemplate(String sessionId, String id) {
        return hashOpsQuizTemplate.get(MessageFormat.format("{0}{1}", INDEX_TEMPLATE_PREFIX, sessionId), id);
    }


}

spring redis template pub/sub and hash 설정