Spring Boot SSE｜SseEmitterの使い方と再接続・nginx設定

Spring MVC で Server-Sent Events（SSE：サーバーからブラウザへ、1本の HTTP 接続で次々にイベントを送る仕組み）を返すには、コントローラーで SseEmitter を返します。 WebFlux なら Flux<ServerSentEvent<T>> を返します。

@GetMapping(path = "/notices", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter notices() {
    SseEmitter emitter = new SseEmitter(30 * 60 * 1000L);   // 30分でタイムアウト
    // 別のスレッドから emitter.send(...) で送る
    return emitter;
}

本番でつまずくのは、ほぼ次の4つです。 どれもこの記事で再現しています。

確認環境：SSE の動きは、SseEmitter と同じ書式（id:・event:・data:）で返す小さなサーバーを Node.js 22.22.2 で作り、curl 8.5.0、Chromium 141.0.7390.37（playwright-core 1.56.1）の EventSource、nginx 1.24.0 を通して確かめました。Spring 側のコードと既定値は、Spring Framework 7.0.9・Spring Boot 4.1.1 のソースとリファレンスで確認しています（確認日 2026年10月13日）。

SSE で流れるデータの形

SSE は、text/event-stream という形式のテキストを、空行で区切って送り続けるだけの仕組みです。 SseEmitter.event() の各メソッドが、次の行を書きます（7.0.9 の SseEmitter のソース）。

SseEmitter.event().id("1").name("notice").data(notice, MediaType.APPLICATION_JSON) は、次のバイト列になります。: のあとに空白は入りません。 HTML の仕様では、: の直後の空白1つは取り除かれるので、data:x と data: x は同じ意味です。

id:1
event:notice
data:{"id":1,"message":"出荷しました"}

curl で、同じ書式のサーバーから受け取った生の応答です（-N でバッファリングを止めています）。

curl -N -i 'http://127.0.0.1:18611/events?count=3&interval=200&retry=3000'

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Transfer-Encoding: chunked

retry:3000

id:1
data:{"id":1,"conn":1}

id:2
data:{"id":2,"conn":1}

id:3
data:{"id":3,"conn":1}

SseEmitter のコード全文（Spring MVC）

通知を全員に配る「ハブ」を1つ作り、接続ごとの SseEmitter を集合で持ちます。 切断・タイムアウト・エラーのときに集合から外すこと、取りこぼしを Last-Event-ID で再送することまで入れています。

package com.example.sse;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class SseApplication {
    public static void main(String[] args) {
        SpringApplication.run(SseApplication.class, args);
    }
}

package com.example.sse;

import java.time.Instant;

public record Notice(long id, String message, Instant at) {
}

package com.example.sse;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicLong;

@Service
public class NoticeHub {

    private static final Logger log = LoggerFactory.getLogger(NoticeHub.class);
    private static final long TIMEOUT_MS = Duration.ofMinutes(30).toMillis();
    private static final int KEEP = 100;                 // 再送用に直近100件を持つ

    private final Set<SseEmitter> emitters = ConcurrentHashMap.newKeySet();
    private final Deque<Notice> recent = new ConcurrentLinkedDeque<>();
    private final AtomicLong seq = new AtomicLong();

    /** 接続を受け付ける。lastEventId はブラウザが再接続時に送ってくる値 */
    public SseEmitter subscribe(String lastEventId) {
        SseEmitter emitter = new SseEmitter(TIMEOUT_MS);
        emitters.add(emitter);

        emitter.onCompletion(() -> emitters.remove(emitter));
        emitter.onTimeout(() -> {               // タイムアウト：こちらから閉じる（ブラウザは再接続する）
            emitters.remove(emitter);
            emitter.complete();
        });
        emitter.onError(e -> emitters.remove(emitter));

        try {
            // 接続直後に1回送る：再接続間隔の指定と、プロキシに応答を早く返すため
            emitter.send(SseEmitter.event().comment("connected").reconnectTime(3000));
            // 取りこぼしの再送
            for (Notice n : since(lastEventId)) {
                emitter.send(toEvent(n));
            }
        } catch (IOException e) {
            // 送信失敗（切断済み）。completeWithError は呼ばない（Javadoc の注記）
            emitters.remove(emitter);
        }
        return emitter;
    }

    /** 全員に配る */
    public Notice publish(String message) {
        Notice n = new Notice(seq.incrementAndGet(), message, Instant.now());
        recent.addLast(n);
        while (recent.size() > KEEP) {
            recent.pollFirst();
        }
        for (SseEmitter emitter : emitters) {
            try {
                emitter.send(toEvent(n));
            } catch (IOException | IllegalStateException e) {
                // IOException：ブラウザが閉じた。IllegalStateException：すでに complete 済み
                emitters.remove(emitter);
            }
        }
        return n;
    }

    /** 15秒ごとのハートビート。切断の検出と、プロキシのタイムアウト対策 */
    @Scheduled(fixedRate = 15_000)
    public void heartbeat() {
        for (SseEmitter emitter : emitters) {
            try {
                emitter.send(SseEmitter.event().comment("keepalive"));
            } catch (IOException | IllegalStateException e) {
                emitters.remove(emitter);
            }
        }
        log.debug("sse connections={}", emitters.size());
    }

    List<Notice> since(String lastEventId) {
        if (lastEventId == null || lastEventId.isBlank()) {
            return List.of();
        }
        long last;
        try {
            last = Long.parseLong(lastEventId);
        } catch (NumberFormatException e) {
            return List.of();
        }
        List<Notice> result = new ArrayList<>();
        for (Notice n : recent) {
            if (n.id() > last) {
                result.add(n);
            }
        }
        return result;
    }

    private SseEmitter.SseEventBuilder toEvent(Notice n) {
        return SseEmitter.event()
                .id(String.valueOf(n.id()))
                .name("notice")
                .data(n, MediaType.APPLICATION_JSON);
    }
}

package com.example.sse;

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@RestController
public class NoticeController {

    private final NoticeHub hub;

    public NoticeController(NoticeHub hub) {
        this.hub = hub;
    }

    @GetMapping(path = "/notices", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter subscribe(@RequestHeader(name = "Last-Event-ID", required = false) String lastEventId) {
        return hub.subscribe(lastEventId);
    }

    @PostMapping("/notices")
    public Notice publish(@RequestBody String message) {
        return hub.publish(message);
    }
}

publish は1件ずつ順に送るので、つながっている人数が多いと遅くなります。数百を超える接続や、複数台のサーバーで配るときは、Redis の Pub/Sub などで各サーバーに届け、各サーバーが自分の接続に送る形にします。

ブラウザ側（EventSource）

<!doctype html>
<meta charset="utf-8">
<title>通知</title>
<ul id="list"></ul>
<script>
  const es = new EventSource("/notices");
  es.addEventListener("notice", (e) => {          // event:notice の行があるイベント
    const n = JSON.parse(e.data);
    const li = document.createElement("li");
    li.textContent = `#${e.lastEventId} ${n.message}`;
    document.getElementById("list").append(li);
  });
  es.onmessage = (e) => console.log("名前なしのイベント", e.data);
  es.onerror = () => console.log("切断。readyState =", es.readyState); // 0 なら再接続中、2 なら終了
</script>

event:notice を付けたイベントは onmessage では受け取れません。 Chromium で確かめると、名前の無いイベントは onmessage に、event:price のイベントは addEventListener("price", …) にだけ届きました。

EventSource の再接続と Last-Event-ID（Chromium で実測）

EventSource は、接続が切れると自動で再接続し、最後に受け取った id を Last-Event-ID ヘッダーで送ってきます。 サーバーが2件送って接続を閉じる、という動きを繰り返させたときの記録です。

39ms    open
241ms   message id=1
443ms   message id=2
447ms   error readyState=0        ← サーバーが閉じた。0 = 再接続を待っている
1456ms  open                      ← retry:1000 の約1秒後に再接続
1654ms  message id=3
1855ms  message id=4
1856ms  error readyState=0
2880ms  open

サーバー側のログでは、再接続のリクエストに Last-Event-ID が付いていました。

#5 GET /events?... Last-Event-ID=-  Accept=text/event-stream
#6 GET /events?... Last-Event-ID=2  Accept=text/event-stream
#7 GET /events?... Last-Event-ID=4  Accept=text/event-stream

条件を変えたときの結果です（Chromium 141）。

サーバーから「もう再接続しないで」と伝えるには、204 No Content を返します（HTML の仕様にもこの方法が書かれています）。逆に、500 などを返すと、一時的な障害でもブラウザは再接続しません。サーバーの再起動中などに 502 を返すプロキシがあると、そこで止まる点に注意します。

SseEmitter に改行を含む文字列を渡すと、7.0.9 のソースでは data: の行に分けて書きます。上の表のとおり、ブラウザ側では元の改行入りの文字列に戻ります。

タイムアウトの決まり方

new SseEmitter()（引数なし）は、Spring MVC の設定、それも無ければサーブレットコンテナの既定値を使います。 Tomcat の非同期処理の既定は 30 秒です。

タイムアウトなしにすると、切断に気づけない接続が残り続けることがあります。 30分などの上限を付け、onTimeout で complete() し、ブラウザの自動再接続と Last-Event-ID でつなぎ直してもらう形が扱いやすくなります。

spring:
  mvc:
    async:
      request-timeout: 30m

切断の検出

Spring Framework のリファレンスには、「Servlet API は、クライアントが去ったことを通知しない」とあります。送ろうとして失敗したときに初めて分かるので、送るものが無い間もハートビートを送ります。HTML の仕様も、プロキシに切られないよう 15秒ごと程度のコメント行を勧めています。

send が IOException で失敗したときは、completeWithError を呼ぶ必要が無いと ResponseBodyEmitter の Javadoc にあります（コンテナからの通知で Spring MVC が後始末をするため）。上のコードでは、集合から外すだけにしています。

WebFlux のコード全文（Flux<ServerSentEvent>）

WebFlux では、Flux<ServerSentEvent<T>> を返します。 ServerSentEvent.builder() の id・event・data・retry・comment が、それぞれの行になります。Spring MVC でも Flux<ServerSentEvent<T>> を返せます（リファレンスの非同期処理の章）。

package com.example.ssewebflux;

import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicLong;

@RestController
public class NoticeStreamController {

    public record Notice(long id, String message, Instant at) {
    }

    private static final int KEEP = 100;

    // 購読者がいないときは捨て、遅い購読者は待たない
    private final Sinks.Many<Notice> sink = Sinks.many().multicast().directBestEffort();
    private final Deque<Notice> recent = new ConcurrentLinkedDeque<>();
    private final AtomicLong seq = new AtomicLong();

    @GetMapping(path = "/notices", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<Notice>> stream(
            @RequestHeader(name = "Last-Event-ID", required = false) String lastEventId) {

        ServerSentEvent<Notice> hello = ServerSentEvent.<Notice>builder()
                .comment("connected")
                .retry(Duration.ofSeconds(3))
                .build();

        Flux<ServerSentEvent<Notice>> replay = Flux.fromIterable(since(lastEventId)).map(this::toSse);
        Flux<ServerSentEvent<Notice>> live = sink.asFlux().map(this::toSse);
        Flux<ServerSentEvent<Notice>> heartbeat = Flux.interval(Duration.ofSeconds(15))
                .map(i -> ServerSentEvent.<Notice>builder().comment("keepalive").build());

        return Flux.concat(Mono.just(hello), replay)
                .concatWith(Flux.merge(live, heartbeat));
        // ブラウザが切断すると購読がキャンセルされ、interval も止まる
    }

    @PostMapping("/notices")
    public Notice publish(@RequestBody String message) {
        Notice n = new Notice(seq.incrementAndGet(), message, Instant.now());
        recent.addLast(n);
        while (recent.size() > KEEP) {
            recent.pollFirst();
        }
        sink.tryEmitNext(n);
        return n;
    }

    private List<Notice> since(String lastEventId) {
        if (lastEventId == null || lastEventId.isBlank()) {
            return List.of();
        }
        long last;
        try {
            last = Long.parseLong(lastEventId);
        } catch (NumberFormatException e) {
            return List.of();
        }
        List<Notice> result = new ArrayList<>();
        for (Notice n : recent) {
            if (n.id() > last) {
                result.add(n);
            }
        }
        return result;
    }

    private ServerSentEvent<Notice> toSse(Notice n) {
        return ServerSentEvent.<Notice>builder()
                .id(String.valueOf(n.id()))
                .event("notice")
                .data(n)
                .build();
    }
}

WebFlux の基本は、関連記事「Spring WebFlux入門｜Mono・Fluxの基本とMVCとの違い」で扱っています。

nginx を通すと届かない・切れる（実測）

nginx の proxy_buffering は既定で on です。 1秒ごとに3件送り、最後に閉じるサーバーを nginx 1.24.0 の後ろに置いて、届いた時刻を測りました。

既定の設定では、3件とも応答が終わった 3.02 秒後にまとめて届きました。 応答が終わらない SSE では、バッファ（既定 4KB か 8KB）がいっぱいになるまで届きません。アプリから X-Accel-Buffering: no を返すと、その応答だけバッファリングが止まります（このヘッダーは nginx が受け取って消し、クライアントには届きませんでした）。

Spring 側でヘッダーを付ける例です。

@GetMapping(path = "/notices", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public ResponseEntity<SseEmitter> subscribe(
        @RequestHeader(name = "Last-Event-ID", required = false) String lastEventId) {
    return ResponseEntity.ok()
            .header("X-Accel-Buffering", "no")
            .header("Cache-Control", "no-cache")
            .body(hub.subscribe(lastEventId));
}

proxy_read_timeout とハートビート

proxy_read_timeout（既定 60 秒）は、上流（アプリ）から何も届かない時間の上限です。3秒に縮めて試しました。

最初のイベントを送るまで何も返さないと、504 になります。 上のコードで、接続直後に comment("connected") を送っているのはこのためです。

SSE 用の location の例です。

location /notices {
    proxy_pass http://app:8080;
    proxy_http_version 1.1;          # 1.29.7 より前の nginx は既定が 1.0
    proxy_set_header Connection "";
    proxy_buffering off;             # アプリ側の X-Accel-Buffering: no でもよい
    proxy_cache off;
    proxy_read_timeout 1h;           # ハートビートの間隔より長く
}

gzip on で text/event-stream も圧縮する設定にしても、proxy_buffering off のもとでは1秒ごとに届きました（nginx 1.24.0 で、届いた分を展開して確認）。nginx のリバースプロキシ全般は、関連記事「Nginxリバースプロキシ設定｜proxy_passの書き方と502/504対処」で扱っています。

よくある質問

Q. WebSocket との使い分けは？ A. SSE はサーバーからの一方向で、ふつうの HTTP なので、認証・プロキシ・ログの仕組みをそのまま使えます。双方向に頻繁にやり取りするなら WebSocket です。関連記事「Spring BootのWebSocket｜STOMPでチャットを作る完成コード」で扱っています。

Q. 同じブラウザでタブをたくさん開くと、つながらなくなります。 A. HTTP/1.1 では、ブラウザが同じサーバーへの同時接続数を制限していて、SSE の接続がその枠を使い続けます。HTML の仕様にもこの注意があります。HTTP/2 で配信すると、1本の接続の中で複数の SSE を流せます。

Q. ヘッダーで認証トークンを送りたい。 A. ブラウザの EventSource は、独自のヘッダーを付けられません。Cookie のセッションを使うか、fetch でストリームを読む実装にします。

Q. LLM の回答を少しずつ返したい。 A. Spring AI の ChatClient の stream() は Flux<String> を返すので、produces = MediaType.TEXT_EVENT_STREAM_VALUE を付けてそのまま返せます。関連記事「Spring AI入門｜ChatClientの使い方とコード全文（2.0対応）」で扱っています。

まとめ

• MVC は SseEmitter、WebFlux は Flux<ServerSentEvent<T>>
• 引数なしの SseEmitter は Tomcat で 30 秒。new SseEmitter(ミリ秒) か spring.mvc.async.request-timeout
• 切断は送信の失敗でしか分からない。15秒ごとのコメント行で検出し、集合から外す
• ブラウザは 自動で再接続し Last-Event-ID を送る（retry なしで約3秒）。204 で再接続を止められる
• nginx は X-Accel-Buffering: no か proxy_buffering off。最初のイベントは接続直後に送る

参考・出典

確認日はいずれも 2026年10月13日です。

• Spring Framework Reference「Asynchronous Requests」（SSE・Disconnects）：https://docs.spring.io/spring-framework/reference/web/webmvc/mvc-ann-async.html
• Spring Framework ソース（タグ v7.0.9、SseEmitter・ResponseBodyEmitter）：https://github.com/spring-projects/spring-framework/tree/v7.0.9/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation
• HTML Living Standard「Server-sent events」：https://html.spec.whatwg.org/multipage/server-sent-events.html
• nginx「ngx_http_proxy_module」（proxy_buffering・proxy_read_timeout・X-Accel-Buffering）：https://nginx.org/en/docs/http/ngx_http_proxy_module.html
• Apache Tomcat 11「The HTTP Connector」（asyncTimeout）：https://tomcat.apache.org/tomcat-11.0-doc/config/http.html
• Spring Boot Reference「Common Application Properties」（spring.mvc.async.request-timeout）：https://docs.spring.io/spring-boot/appendix/application-properties/index.html

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。