Consumer Group으로 메시지를 읽을 때, 서버는 메시지를 받았다는걸 기억하고 PEL(Pending Entries List)에 추가합니다. 전달되었지만 아직 확인 되는 메시지 목록을 제거 하려면 StreamOperations.acknowledge 통해 메시지를 확인해야 합니다.
StreamMessageListenerContainer<String,
MapRecord<String, String, String>
> container = ...
container.receive(Consumer.from("my-group", "my-consumer"),
StreamOffset.create("my-stream", ReadOffset.lastConsumed()),
msg -> {
//...
redisTemplate.opsForStream().acknowledge("my-group", msg);
}
)
Sream 읽기는 offset 사양을 지정된 오프셋의 메시지를 사용합니다. ReadOffset 읽기 오프셋 사양을 나타냅니다. Redis는 스트림 독립 실행형으로 소비하는지 소비자 그룹 내에서 소비하는지에 따라 3가지 오프셋 변형을 지원합니다.
MessageContainer 기반 소비의 문맥은, 메시지를 소비할 때 읽기 offset을 증가해야 합니다. 진행된 요청된 ReadOffset 및 consume mode (consume group에 포함되는지)에 따라 달라집니다.
특정 메시지 ID와 마지막으로 사용된 메시지에서 읽는 것은 스트림에 추가된 모든 메시지의 사용을 보장하는 안전한 작업으로 간주될 수 있습니다. 읽기에 최신 메ㅣ지를 사용하면 풀작업이 데드 타임 상태에 있는 동안 스트림에 추가된 메시지를 건너뛸 수 있습니다. 폴링은 개별 폴링 사이에 메시지가 도착할 수 있는 데드타임을 도입합니다. 스트림 소비는 선형 연속 읽기가 아니라 XREAD 호출로 분할됩니다.
스트림으로 전송된 모든 레코드는 바이너리 형식으로 직렬화되어야 합니다. 스트림이 해시 데이터구조에 가깝기 때문에 스트림 키, 필드 이름 및 값은 구성된 직렬 변환기를 사용합니다. RedisTemplate