[Spring Data Redis] Streams

 

1. Streams 란

 

  - 이벤트 저장소

 

 

* 다음 게시글 참조

https://kjwit.tistory.com/entry/Redis-Streams

[Redis] Streams

 

 

 

2. Streams  사용 방법 

 

 - chatRoomEventListener 생성

 

public class ChatRoomEventListener implements StreamListener<String, MapRecord<String, String, String>> {

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        Map<String, String> map = message.getValue();
    }
}

 

 

 

- chatRoomService 생성 ( 알림 생성 역할 )

 

@Service
@RequiredArgsConstructor
public class ChatService {
 
    private final RedisTemplate<String,String> redisTemplate;
    
    public void sendMessage(String chatRoomName, String userId, String msg){

        Map<String, String> fieldMap = new HashMap<>();
        fieldMap.put("roomId", chatRoomName);
        fieldMap.put("userId", userId);
        fieldMap.put("massage", msg);
        redisTemplate.opsForStream().add("chat-room-events", fieldMap);
    }
}

 

 

 

- redis 에서 Stream 생성 

 

XGROUP CREATE chat-room-events chat-room-service-group $ MKSTREAM

 

 

 

 - redisConfig 에 다음을 추가

 

@Bean
    public Subscription subscription(RedisConnectionFactory factory){
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(1))
                .build();
        StreamMessageListenerContainer listenerContainer = StreamMessageListenerContainer.create(factory, options);

        Subscription subscription = listenerContainer.receiveAutoAck(Consumer.from("chat-room-service-group", "instance-1"), StreamOffset.create("chat-room-events", ReadOffset.lastConsumed()), chatRoomEventListener);

        listenerContainer.start();

        return subscription;
    }