Subscription과 느린 소비자 정책
query와 mutation은 요청이 끝나면 연결도 끝납니다. subscription은 한 번 열린 연결에 세션 완료 event를 지속적으로 보냅니다. Spring GraphQL의 HTTP endpoint는 Accept: text/event-stream인 subscription POST를 SSE로 처리하므로 WebSocket을 추가하지 않고도 스트리밍을 학습할 수 있습니다.
Subscription은 오래 유지되는 스트림 계약이다
여기서 가장 중요한 것은 “상태 변경이 완료된 뒤에만 event를 발행한다”는 순서입니다. DB transaction이 rollback됐는데 event를 먼저 보내면 구독자는 존재하지 않는 상태를 보게 됩니다. 또한 생산자보다 느린 구독자를 무제한 queue로 버티게 하면 memory가 끝없이 늘어납니다. 이 예제는 느린 구독자의 event를 버리고 빠른 구독자를 보호하는 directBestEffort 정책을 선택합니다.
완료 이벤트 Flux를 GraphQL에 연결한다
Sinks.Many는 imperative service 결과를 Reactive Streams Flux로 넘기는 bridge입니다. sink를 전역 event store로 오해하면 안 됩니다. 프로세스가 재시작하면 이벤트는 사라지고, 구독하지 않은 동안의 이벤트도 재생되지 않습니다. 지속적 전달이 필요하면 7장의 Outbox와 broker같은 다른 장치가 필요합니다.
HTTP SSE subscription은 ch6-1에서 추가한 GraphQL starter와 WebMVC transport로 처리하므로 WebSocket starter나 별도 endpoint 설정을 추가하지 않습니다.
type SessionCompletedEvent {
id: ID!
subject: String!
}
type Subscription {
sessionCompleted: SessionCompletedEvent!
}package com.andongmin.studylog.graphql;
import java.util.UUID;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
@Component
public class SessionEventPublisher {
private final Sinks.Many<SessionCompletedEvent> completed =
Sinks.many().multicast().directBestEffort();
public void publish(SessionCompletedEvent event) {
Sinks.EmitResult result = completed.tryEmitNext(event);
if (result == Sinks.EmitResult.FAIL_NON_SERIALIZED
|| result == Sinks.EmitResult.FAIL_TERMINATED) {
throw new IllegalStateException("subscription event was not delivered: " + result);
}
}
public Flux<SessionCompletedEvent> events() {
return completed.asFlux();
}
public record SessionCompletedEvent(UUID id, String subject) {
}
}ch6-2의 StudySessionMutationController.java를 아래 코드로 교체합니다. service.complete()의 transaction이 commit된 뒤 Controller가 완료 이벤트를 발행하므로 활성 구독이 실제 상태 전이와 연결됩니다.
package com.andongmin.studylog.graphql;
import java.time.LocalDate;
import java.util.UUID;
import jakarta.validation.Valid;
import jakarta.validation.constraints.Max;
import jakarta.validation.constraints.Min;
import jakarta.validation.constraints.NotBlank;
import org.springframework.graphql.data.method.annotation.Argument;
import org.springframework.graphql.data.method.annotation.MutationMapping;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.stereotype.Controller;
import org.springframework.validation.annotation.Validated;
import com.andongmin.studylog.session.CreateStudySessionRequest;
import com.andongmin.studylog.session.StudySessionResponse;
import com.andongmin.studylog.session.VersionedStudySessionService;
@Controller
@Validated
public class StudySessionMutationController {
private final VersionedStudySessionService service;
private final SessionEventPublisher events;
public StudySessionMutationController(
VersionedStudySessionService service, SessionEventPublisher events) {
this.service = service;
this.events = events;
}
@MutationMapping
@PreAuthorize("hasAuthority('SCOPE_study.write')")
public StudySessionResponse createStudySession(@Argument @Valid CreateInput input) {
return service.create(new CreateStudySessionRequest(
input.subject(), input.topic(), LocalDate.parse(input.studyDate()),
input.minutes()));
}
@MutationMapping
@PreAuthorize("hasAuthority('SCOPE_study.write')")
public StudySessionResponse completeStudySession(@Argument UUID id) {
StudySessionResponse completed = service.complete(id);
events.publish(new SessionEventPublisher.SessionCompletedEvent(
completed.id(), completed.subject()));
return completed;
}
public record CreateInput(
@NotBlank String subject,
@NotBlank String topic,
@NotBlank String studyDate,
@Min(1) @Max(1440) int minutes) {
}
}package com.andongmin.studylog.graphql;
import org.springframework.graphql.data.method.annotation.SubscriptionMapping;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;
@Controller
public class StudySessionSubscriptionController {
private final SessionEventPublisher events;
public StudySessionSubscriptionController(SessionEventPublisher events) {
this.events = events;
}
@SubscriptionMapping
@PreAuthorize("hasAuthority('SCOPE_study.read')")
public Flux<SessionEventPublisher.SessionCompletedEvent> sessionCompleted() {
return events.events();
}
}두 클라이언트의 구독과 해제를 확인한다
서버, SSE 구독, mutation 요청을 서로 다른 terminal에 놓습니다. 먼저 read token으로 구독을 열어 놓은 뒤, write token으로 세션을 생성하고 그 ID를 완료합니다. 완료 response가 돌아온 후 SSE terminal에 event 한 건이 나와야 transaction과 발행 순서가 맞은 것입니다. 구독을 닫은 후에는 새 event를 받지 않습니다.
export LOCAL_JWT_SECRET="study-log-local-development-key-2026"
SPRING_PROFILES_ACTIVE=local ./mvnw -q spring-boot:runexport LOCAL_JWT_SECRET="study-log-local-development-key-2026"
READ_TOKEN="$(java dev/LocalToken.java "$LOCAL_JWT_SECRET" "study.read")"
curl -N http://localhost:8080/graphql \
-H "Authorization: Bearer $READ_TOKEN" \
-H "Accept: text/event-stream" -H "Content-Type: application/json" \
-d "{\"query\":\"subscription { sessionCompleted { id subject } }\"}"구독을 연 채 별도 terminal에서 생성과 완료 mutation을 차례로 보냅니다.
export LOCAL_JWT_SECRET="study-log-local-development-key-2026"
WRITE_TOKEN="$(java dev/LocalToken.java "$LOCAL_JWT_SECRET" "study.write")"
CREATE_RESPONSE="$(curl -s http://localhost:8080/graphql \
-H "Authorization: Bearer $WRITE_TOKEN" \
-H "Content-Type: application/json" --data-binary @create-session.json)"
ID="$(printf '%s' "$CREATE_RESPONSE" | \
sed -n 's/.*"id":"\([^"]*\)".*/\1/p')"
test -n "$ID"
curl -s http://localhost:8080/graphql \
-H "Authorization: Bearer $WRITE_TOKEN" \
-H "Content-Type: application/json" \
-d "{\"query\":\"mutation { completeStudySession(id: \\\"$ID\\\") { id status } }\"}"study.read token으로 구독을 연결한 뒤 별도 terminal에서 study.write token으로 완료 mutation을 보내면 이벤트 한 건이 전달되고, 연결 해제 뒤에는 subscriber가 제거되어야 합니다.유실과 지연 중 무엇을 허용할지 정한다
directBestEffort 정책은 느린 소비자에게 유실을 허용하고 전체 지연을 줄입니다. 모든 event가 반드시 필요하다면 sink의 buffer만 크게 만들 것이 아니라, 재생 가능한 broker와 offset·acknowledgement 계약을 선택해야 합니다. 실시간 UI 알림처럼 최신 상태가 중요한지, 결제·감사 이력처럼 한 건도 잃으면 안 되는지로 선택이 달라집니다.
지금까지는 하나의 프로세스 안에서 REST·GraphQL·event를 공유했습니다. 프로세스 경계를 나누면 method 호출은 HTTP와 message로 바뀌고, transaction과 실패의 의미도 달라집니다. 7장에서 먼저 업무와 데이터 소유권으로 경계부터 잡습니다.