Redis Streams 순서 보장 가이드

← 기술 결정 목록

결정: Redis Streams (초기) → Kafka (확장 시) 상태: 승인됨

목차

1. Redis Streams 순서 보장 특성

1.1 기본 순서 보장

Redis Streams는 기본적으로 순서를 보장합니다.

특징:

예시:

1
2
3
4
5
6
7
8
9

Producer가 메시지를 보낸 순서:
1. Message A
2. Message B  
3. Message C

Consumer가 읽는 순서:
1. Message A ✅
2. Message B ✅
3. Message C ✅

1.2 컨슈머 그룹 내 순서 보장

단일 컨슈머 그룹:

예시:

1
2
3
4
5

// 컨슈머 그룹 "message-processors" 생성
XGROUP CREATE chat:messages message-processors 0

// 메시지 읽기 (순서 보장)
XREADGROUP GROUP message-processors consumer-1 COUNT 10 STREAMS chat:messages >

결과:

1.3 여러 컨슈머가 있을 때

같은 컨슈머 그룹 내 여러 컨슈머:

예시:

1
2
3
4

Stream: [A, B, C, D, E, F]

Consumer-1: A, C, E (순서대로 처리)
Consumer-2: B, D, F (순서대로 처리)

주의사항:

2. Co-Talk 프로젝트에서의 활용

2.1 메시지 저장 시나리오

사용 사례:

순서 보장 필요성:

2.2 구현 방법

방법 1: 채팅방별 Stream (권장)

1
2
3
4
5
6
7
8
9

// 각 채팅방마다 별도 Stream 생성
String streamKey = "chat:messages:" + chatRoomId;

// 메시지 추가 (순서 보장)
redisTemplate.opsForStream().add(streamKey, Map.of(
    "senderId", senderId,
    "content", content,
    "timestamp", System.currentTimeMillis()
));

장점:

방법 2: 단일 Stream + Consumer Group

1
2
3
4
5
6
7
8
9
10
11
12
13
14

// 단일 Stream에 모든 메시지 저장
String streamKey = "chat:messages";

// 메시지 추가 (순서 보장)
redisTemplate.opsForStream().add(streamKey, Map.of(
    "chatRoomId", chatRoomId,
    "senderId", senderId,
    "content", content
));

// 컨슈머 그룹으로 읽기
XReadGroupArgs args = XReadGroupArgs.StreamOffset.from(streamKey, ">");
List<MapRecord<String, Object, Object>> messages = 
    redisTemplate.opsForStream().readGroup(consumerGroup, consumerName, args);

장점:

단점:

2.3 권장 구현: 채팅방별 Stream

이유:

구현:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

@Service
public class MessageQueueService {
    
    public void enqueueMessage(String chatRoomId, Message message) {
        String streamKey = "chat:messages:" + chatRoomId;
        
        // 메시지 추가 (순서 보장)
        redisTemplate.opsForStream().add(streamKey, Map.of(
            "id", message.getId(),
            "senderId", message.getSenderId(),
            "content", message.getContent(),
            "timestamp", message.getTimestamp()
        ));
    }
    
    @Scheduled(fixedDelay = 1000)
    public void processMessages() {
        // 각 채팅방의 Stream에서 메시지 읽기
        // 순서대로 처리됨
    }
}

3. 트래픽 수준별 순서 보장

3.1 낮은 트래픽 (초기)

특징:

순서 보장:

3.2 중간 트래픽

특징:

순서 보장:

3.3 높은 트래픽

특징:

순서 보장:

최적화:

4. Redis Streams vs Kafka 순서 보장 비교

4.1 순서 보장

특성 Redis Streams Kafka
기본 순서 보장 ✅ 보장 ✅ 보장
트래픽과 무관 ✅ 보장 ✅ 보장
파티션 내 순서 ✅ 보장 ✅ 보장
여러 파티션 ⚠️ 주의 필요 ✅ 파티션 키로 보장

4.2 차이점

Redis Streams:

Kafka:

4.3 Co-Talk 프로젝트에서

Redis Streams로 충분한 이유:

Kafka로 전환할 때:

5. 실제 구현 예시

5.1 메시지 저장 (순서 보장)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

@Service
public class MessageService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    public void sendMessage(String chatRoomId, Message message) {
        // 1. WebSocket으로 즉시 전달
        websocketService.sendToRoom(chatRoomId, message);
        
        // 2. Redis Streams에 저장 (순서 보장)
        String streamKey = "chat:messages:" + chatRoomId;
        redisTemplate.opsForStream().add(streamKey, Map.of(
            "id", message.getId(),
            "senderId", message.getSenderId(),
            "content", message.getContent(),
            "timestamp", message.getTimestamp()
        ));
    }
}

5.2 메시지 처리 (순서 보장)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

@Service
public class MessageProcessor {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private MessageRepository messageRepository;
    
    @Scheduled(fixedDelay = 1000)
    public void processMessages() {
        // 모든 채팅방의 Stream에서 메시지 읽기
        Set<String> chatRoomIds = getActiveChatRooms();
        
        for (String chatRoomId : chatRoomIds) {
            String streamKey = "chat:messages:" + chatRoomId;
            String consumerGroup = "message-processors";
            String consumerName = "consumer-1";
            
            // 메시지 읽기 (순서 보장)
            List<MapRecord<String, Object, Object>> messages = 
                redisTemplate.opsForStream().readGroup(
                    consumerGroup, 
                    consumerName,
                    XReadGroupArgs.StreamOffset.from(streamKey, ">")
                );
            
            // 순서대로 처리
            for (MapRecord<String, Object, Object> record : messages) {
                processMessage(record);
                // ACK 처리
                redisTemplate.opsForStream().acknowledge(
                    consumerGroup, 
                    streamKey, 
                    record.getId()
                );
            }
        }
    }
    
    private void processMessage(MapRecord<String, Object, Object> record) {
        // PostgreSQL에 저장 (순서대로)
        Message message = convertToMessage(record);
        messageRepository.save(message);
    }
}

6. 주의사항 및 최적화

6.1 주의사항

여러 컨슈머가 있을 때:

컨슈머 그룹 설정:

6.2 최적화

배치 처리:

1
2
3

// 여러 메시지를 한 번에 읽기
XReadGroupArgs args = XReadGroupArgs.StreamOffset.from(streamKey, ">")
    .count(100); // 최대 100개씩 읽기

ACK 처리:

1
2
3
4
5
6

// 메시지 처리 후 즉시 ACK
redisTemplate.opsForStream().acknowledge(
    consumerGroup, 
    streamKey, 
    record.getId()
);

PENDING 메시지 처리:

1
2
3
4
5
6
7

// 처리되지 않은 메시지 재처리
List<MapRecord<String, Object, Object>> pendingMessages = 
    redisTemplate.opsForStream().pending(
        consumerGroup, 
        streamKey, 
        consumerName
    );

7. 결론

✅ Redis Streams 순서 보장

핵심:

Co-Talk 프로젝트:

최종 권장: