How Redis Pub/Sub Works

struct redisServer {
    dict *pubsub_channels;  // 채널명 → 구독자 리스트 해시테이블
    list *pubsub_patterns;  // 패턴 구독자들 링크드리스트
    // ...
}

// 브라우저에서 전송
stompClient.send('/app/room/ABC123/update-ready', {}, JSON.stringify({
  joinCode: "ABC2",
  playerName: "홍길동",
  isReady: true
}));

@MessageMapping("/room/{joinCode}/update-ready")
public void broadcastReady(@DestinationVariable String joinCode, 
                          ReadyChangeMessage message) {
    final PlayerReadyEvent event = new PlayerReadyEvent(
        joinCode, 
        message.playerName(),
        message.isReady()
    );
    roomEventPublisher.publishEvent(event);
}

public <T extends RoomBaseEvent> void publishEvent(T event) {
    redisTemplate.convertAndSend(roomEventTopic.getTopic(), event);
}

// RedisTemplate 내부 동작
public void convertAndSend(String channel, Object message) {
    // 1. 채널명 직렬화: "room.events" → byte[]
    byte[] rawChannel = "room.events".getBytes(StandardCharsets.UTF_8);
    
    // 2. 메시지 객체 직렬화: PlayerReadyEvent → JSON → byte[]
    GenericJackson2JsonRedisSerializer serializer = ...;
    byte[] rawMessage = serializer.serialize(event);
    // 결과: {"eventId":"uuid-123","joinCode":"ABC2","playerName":"홍길동","isReady":true,...}
    
    // 3. Lettuce Connection으로 전달
    connection.publish(rawChannel, rawMessage);
}

*3\r\n
$7\r\n
PUBLISH\r\n
$11\r\n
room.events\r\n
$152\r\n
{"eventId":"uuid-123","joinCode":"ABC2","playerName":"홍길동","isReady":true,...}\r\n

// Lettuce 내부
public Long publish(byte[] channel, byte[] message) {
    // RESP 프로토콜로 인코딩
    CommandArgs<byte[], byte[]> args = new CommandArgs<>(codec)
        .addKey(channel)   // room.events
        .addValue(message); // JSON 메시지
    
    Command<byte[], byte[], Long> command = commandBuilder.publish(args);
    
    // Netty를 통해 비동기 전송
    return dispatch(command);
}

// Netty 내부 (간략화)
Channel channel = getChannel(); // Redis 서버와의 TCP 연결

// ByteBuf에 RESP 데이터 작성
ByteBuf buffer = channel.alloc().buffer();
buffer.writeBytes(respProtocolBytes);

// 비동기 전송
ChannelFuture future = channel.writeAndFlush(buffer);

// 결과를 기다리지 않고 즉시 리턴 (Non-blocking)
future.addListener(writeCompleteListener);

// Redis 이벤트 루프 (간략화)
void aeMain(aeEventLoop *eventLoop) {
    while (!eventLoop->stop) {
        // epoll_wait() 호출 - 이벤트 발생 시까지 대기
        numevents = aeApiPoll(eventLoop, tvp);
        
        for (int j = 0; j < numevents; j++) {
            aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
            
            // 읽기 이벤트 (클라이언트가 데이터 전송)
            if (fe->mask & AE_READABLE) {
                fe->rfileProc(eventLoop, fd, fe->clientData, mask);
            }
            
            // 쓰기 이벤트 (클라이언트에게 전송 가능)
            if (fe->mask & AE_WRITABLE) {
                fe->wfileProc(eventLoop, fd, fe->clientData, mask);
            }
        }
    }
}

// 1. 소켓에서 데이터 읽기 (간략화)
void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
    client *c = (client*) privdata;
    
    // TCP 소켓에서 데이터 읽기
    nread = read(fd, c->querybuf + qblen, readlen);
    
    // RESP 프로토콜 파싱
    processInputBuffer(c);
}

// 2. PUBLISH 명령 실행 (간략화)
void publishCommand(client *c) {
    // c->argv[1] = "room.events" (채널명)
    // c->argv[2] = JSON 메시지
    
    int receivers = 0;
    
    // pubsub_channels 해시테이블에서 채널 조회 (O(1))
    dictEntry *de = dictFind(server.pubsub_channels, c->argv[1]);
    
    if (de) {
        list *subscribers = dictGetVal(de);
        
        // 구독자 리스트 순회 (O(N), N=구독자 수)
        listNode *ln;
        listIter li;
        listRewind(subscribers, &li);
        
        while ((ln = listNext(&li)) != NULL) {
            client *subscriber = ln->value;
            
            // 각 구독자의 output buffer에 메시지 추가
            addReplyPubsubMessage(subscriber, c->argv[1], c->argv[2]);
            receivers++;
        }
    }
    
    // 발행자에게 응답: 몇 명에게 전송했는지
    addReplyLongLong(c, receivers);
}

// 3. output buffer의 데이터를 실제로 전송 (간략화)
void writeToClient(int fd, client *c, int handler_installed) {
    while (clientHasPendingReplies(c)) {
        // write() 시스템콜로 TCP 소켓에 전송
        nwritten = write(fd, c->buf + c->sentlen, c->bufpos - c->sentlen);
        
        if (nwritten <= 0) break;
        
        c->sentlen += nwritten;
    }
}

@Override
public void onMessage(Message message, byte[] pattern) {
    // 1. 메시지 body를 String으로 변환
    String body = new String(message.getBody());
    
    // 2. eventType 추출
    RoomEventType eventType = extractEventType(body); // PLAYER_READY
    
    // 3. JSON을 Java 객체로 역직렬화
    PlayerReadyEvent event = objectMapper.readValue(body, PlayerReadyEvent.class);
    
    // 4. 핸들러로 위임
    RoomEventHandler handler = handlerFactory.getHandler(eventType);
    handler.handle(event);
}

@Override
public void handle(PlayerReadyEvent event) {
    // 로컬 메모리에서 Room 조회 (네트워크 통신 없음)
    Room room = roomQueryService.getByJoinCode(new JoinCode(event.joinCode()));
    
    // Player 찾기
    Player player = room.findPlayer(new PlayerName(event.playerName()));
    
    // 상태 변경 (로컬 메모리)
    player.updateReadyState(event.isReady());
    
    // 로컬 캐시 업데이트
    roomCommandService.save(room);
    
    // WebSocket 브로드캐스팅
    List<PlayerResponse> responses = room.getPlayers().stream()
        .map(PlayerResponse::from)
        .toList();
    
    messagingTemplate.convertAndSend(
        "/topic/room/" + event.joinCode(),
        WebSocketResponse.success(responses)
    );
}

stompClient.subscribe('/topic/room/ABC123', (message) => {
    const response = JSON.parse(message.body);
    // {success: true, data: [{playerName: "홍길동", isReady: true}, ...]}
    
    updatePlayerList(response.data);
});