세션 레지스트리와 브로드캐스트

채팅 서버의 핵심은 소켓을 여러 개 받는 일이 아니라 현재 활성 연결 집합을 알고 있다는 데 있습니다. 한 세션이 받은 메시지를 다른 세션으로 보내려면 서버가 각 연결의 전송 경계를 보관해야 합니다. 이 집합은 접속과 퇴장이 동시에 일어나는 동안에도 일관성을 유지해야 하고, 서버 종료 시에는 남아 있는 연결을 찾아 닫을 수 있어야 합니다.

세션 관리자는 단순한 List 포장이 아닙니다. 등록 허용 여부, 중복 등록, 스냅샷 생성, 전송 실패 격리, 전체 종료가 한 수명 모델을 이룹니다. 특히 컬렉션 잠금을 잡은 채 네트워크 쓰기를 수행하면 한 명의 느린 클라이언트가 입장과 퇴장까지 막습니다. 공유 상태 보호와 외부 I/O 실행을 분리하는 것이 서버 확장성의 첫 경계입니다.

잠금 안의 소켓 전송

아래 구현은 synchronized 덕분에 목록 자체는 깨지지 않습니다. 그러나 broadcast가 세션 수만큼 네트워크 쓰기를 수행하는 동안 같은 모니터를 계속 소유합니다. 어느 한 send가 운영체제 송신 버퍼에서 오래 멈추면 새 세션의 add, 끊어진 세션의 remove, 종료 스레드의 스냅샷이 모두 입구에서 대기합니다.

bad/LockedBroadcastDirectory.java

import java.util.ArrayList;
import java.util.List;

public final class LockedBroadcastDirectory {
    interface Session {
        void send(String message) throws Exception;
    }

    private final List<Session> sessions = new ArrayList<>();

    synchronized void add(Session session) {
        sessions.add(session);
    }

    synchronized void remove(Session session) {
        sessions.remove(session);
    }

    synchronized void broadcast(String message) throws Exception {
        for (Session session : sessions) {
            session.send(message);
        }
    }

    public static void main(String[] args) {
        System.out.println("compile-only lock scope counterexample");
    }
}

잠금 시간을 줄이려면 현재 대상의 불변 스냅샷을 잠금 안에서 만들고, 실제 전송은 잠금 밖에서 실행합니다. 이때 스냅샷에 포함된 세션이 직후 종료될 수 있다는 점을 받아들여야 합니다. 전송 시점에 닫혔다면 해당 대상만 실패로 기록하고 나머지 참가자 전송을 계속합니다. “모든 대상이 정확히 같은 순간에 열려 있어야 한다”는 보장은 현실적인 네트워크 브로드캐스트 계약이 아닙니다.

또 다른 함정은 첫 전송 실패에서 루프를 끝내는 것입니다. 앞쪽 세션 하나의 오류 때문에 뒤쪽 정상 세션이 메시지를 받지 못합니다. 서버는 대상별 결과를 모아 성공 수와 실패 ID를 남기고, 실패한 세션을 닫거나 제거할지 별도 정책으로 판단해야 합니다. 메시지 전달 보장을 더 높이려면 각 세션에 유한 출력 큐를 두어 브로드캐스트 스레드는 큐 삽입만 수행하게 만들 수 있습니다.

세션 레지스트리

다음 구현은 등록 상태를 한 모니터로 보호하고, sendAll에서는 복사한 세션 목록만 순회합니다. 반환되는 BroadcastResult는 호출자가 일부 실패를 관찰할 수 있게 합니다. 실패를 단순 로그로 삼키면 운영자가 “메시지가 몇 명에게 전달되었는가”를 알 수 없으므로 결과 타입을 서버 계층까지 전달합니다.

src/SessionDirectory.java

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

public final class SessionDirectory {
    interface Endpoint {
        String id();

        void send(String message) throws Exception;

        void close();
    }

    record DeliveryFailure(String sessionId, String cause) {
    }

    record BroadcastResult(int attempted, int delivered,
                           List<DeliveryFailure> failures) {
        BroadcastResult {
            failures = List.copyOf(failures);
        }
    }

    private final Map<String, Endpoint> sessions = new LinkedHashMap<>();
    private boolean closing;

    synchronized boolean register(Endpoint endpoint) {
        if (closing || sessions.containsKey(endpoint.id())) {
            return false;
        }
        sessions.put(endpoint.id(), endpoint);
        return true;
    }

    synchronized boolean remove(String id, Endpoint expected) {
        return sessions.remove(id, expected);
    }

    synchronized List<Endpoint> snapshot() {
        return List.copyOf(sessions.values());
    }

    BroadcastResult sendAll(String message) {
        List<Endpoint> targets = snapshot();
        List<DeliveryFailure> failures = new ArrayList<>();
        int delivered = 0;

        for (Endpoint target : targets) {
            try {
                target.send(message);
                delivered += 1;
            } catch (Exception error) {
                failures.add(new DeliveryFailure(
                        target.id(),
                        error.getClass().getSimpleName()));
            }
        }
        return new BroadcastResult(targets.size(), delivered, failures);
    }

    List<Endpoint> beginClose() {
        synchronized (this) {
            if (closing) {
                return List.of();
            }
            closing = true;
            List<Endpoint> targets = List.copyOf(sessions.values());
            sessions.clear();
            return targets;
        }
    }

    public static void main(String[] args) {
        var directory = new SessionDirectory();
        List<String> received = new ArrayList<>();

        directory.register(endpoint("a", received, false));
        directory.register(endpoint("b", received, true));
        directory.register(endpoint("c", received, false));

        BroadcastResult result = directory.sendAll("hello");
        System.out.println(result);
        System.out.println(received);
        System.out.println("closing=" + directory.beginClose().size());
        System.out.println("late=" + directory.register(
                endpoint("late", received, false)));
    }

    private static Endpoint endpoint(
            String id,
            List<String> received,
            boolean fail) {
        return new Endpoint() {
            @Override
            public String id() {
                return id;
            }

            @Override
            public void send(String message) {
                if (fail) {
                    throw new IllegalStateException("closed");
                }
                received.add(id + ":" + message);
            }

            @Override
            public void close() {
                received.add(id + ":closed");
            }
        };
    }
}

beginClose()는 목록을 비운 뒤 외부에서 각 엔드포인트를 닫을 수 있도록 대상만 반환합니다. close 콜백은 임의의 시간 동안 멈추거나 다시 레지스트리를 호출할 수 있으므로 모니터 안에서 실행하지 않습니다. closing이 참이 된 뒤에는 늦게 도착한 등록을 거절합니다. 그렇지 않으면 종료 스냅샷을 만든 직후 새 세션이 추가되어 서버 프로세스를 붙잡는 유령 연결이 생깁니다.

ID 중복 정책도 명시적이어야 합니다. 연결 ID는 서버가 생성한 불변 식별자를 쓰고, 사용자가 바꾸는 표시 이름과 분리합니다. 이름을 키로 사용하면 /change가 컬렉션 키 교체와 브로드캐스트 표시를 동시에 수행해야 하고 동일 이름 경쟁도 처리하기 어렵습니다. 세션 ID는 수명 추적용, 사용자 이름은 채팅 상태용이라는 두 역할을 유지합니다.

연결 소유권 인계

ServerSocket.accept()가 반환한 Socket은 아직 어느 세션 작업에도 속하지 않습니다. 서버가 세션 객체를 생성하고 레지스트리에 등록한 뒤 실행기에 제출하면 성공 시 소유권은 세션 run으로 넘어갑니다. 각 중간 단계에서 실패할 수 있으므로 보상 동작을 표로 설계해야 합니다.

실패 지점	레지스트리 상태	소켓을 닫을 주체	필요한 보상
세션 생성 중 오류	미등록	`accept` 루프	즉시 `close`
서버 종료로 등록 거부	미등록	`accept` 루프	`close` 후 다음 `accept` 중지
등록 뒤 제출 거부	등록됨	제출 코드	`remove` 후 `close`
제출 성공 뒤 실행 오류	등록됨	세션 `finally`	`remove`와 `close`

제출을 먼저 하고 등록하면 매우 빠른 작업이 완료되어 remove를 호출한 뒤, accept 스레드가 이미 끝난 세션을 목록에 추가할 수 있습니다. 반대로 등록 후 제출에서는 제출 실패를 반드시 보상해야 합니다. 어느 순서를 택해도 실패 분기가 존재하며, 등록→제출→실행 인계가 이해하기 쉬운 편입니다.

src/ManagedChatSession.java

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

public final class ManagedChatSession implements Runnable, AutoCloseable {
    interface Registry {
        boolean add(ManagedChatSession session);

        void remove(ManagedChatSession session);
    }

    private final String id;
    private final Registry registry;
    private final List<String> inbound;
    private final List<String> handled;
    private final AtomicBoolean closed = new AtomicBoolean();

    ManagedChatSession(
            String id,
            Registry registry,
            List<String> inbound,
            List<String> handled) {
        this.id = id;
        this.registry = registry;
        this.inbound = List.copyOf(inbound);
        this.handled = handled;
    }

    boolean admit() {
        return registry.add(this);
    }

    @Override
    public void run() {
        try {
            for (String frame : inbound) {
                if (closed.get()) {
                    return;
                }
                handled.add(id + "=" + frame);
            }
        } finally {
            close();
        }
    }

    @Override
    public void close() {
        if (!closed.compareAndSet(false, true)) {
            return;
        }
        registry.remove(this);
        handled.add(id + "=closed");
    }

    public static void main(String[] args) {
        List<ManagedChatSession> active = new ArrayList<>();
        Registry registry = new Registry() {
            @Override
            public boolean add(ManagedChatSession session) {
                return active.add(session);
            }

            @Override
            public void remove(ManagedChatSession session) {
                active.remove(session);
            }
        };

        List<String> handled = new ArrayList<>();
        var session = new ManagedChatSession(
                "s-1",
                registry,
                List.of("/join|Mina", "/message|hi"),
                handled);

        System.out.println("admitted=" + session.admit());
        session.run();
        session.close();
        System.out.println("active=" + active.size());
        System.out.println(handled);
    }
}

이 예제의 레지스트리는 단일 스레드 목록이지만 세션의 멱등 close 계약을 확인하기에는 충분합니다. 운영 구현에서는 add와 remove를 같은 동기화 정책으로 보호하고, handled 같은 가변 결과 수집기도 동시 접근에 맞는 구조를 써야 합니다. 중요한 점은 정상 완료와 외부 종료가 동시에 close()를 호출해도 제거 콜백이 한 번뿐이라는 것입니다.

느린 수신자 격리

스냅샷 밖 전송은 레지스트리 잠금 문제를 해결하지만 브로드캐스트 호출 자체는 가장 느린 대상의 합계 시간만큼 걸립니다. 참가자 수가 많거나 모바일 연결이 불안정하면 세션별 유한 발신 큐와 단일 기록기를 둡니다. 브로드캐스트는 큐에 프레임을 넣고 즉시 다음 대상으로 이동하며, 각 기록기가 자기 소켓에 순서대로 씁니다.

큐가 가득 찼을 때 무제한 대기하면 다시 전체 브로드캐스트가 멈춥니다. 오래된 일반 메시지 삭제, 새 메시지 거절, 느린 세션 종료 중 서비스 의미에 맞는 정책을 골라야 합니다. 입장·퇴장 같은 제어 메시지와 대화 메시지의 중요도가 다르다면 큐를 나누거나 우선순위를 줄 수 있지만, 순서가 뒤집히지 않는지 별도 검토가 필요합니다.

전송 지표에는 큐 깊이, 큐 삽입 거부, 실제 쓰기 지연, 마지막 성공 시각을 포함합니다. 브로드캐스트 호출 성공은 소켓 전송 완료가 아니라 각 세션 큐에 수락되었다는 뜻일 수 있으므로 메서드 이름과 반환 타입에 전달 단계를 드러냅니다. 강한 전달 보장이 필요하다면 클라이언트 확인 응답과 메시지 ID가 추가되어야 하며, 그것은 현재의 단순 TCP 채팅보다 높은 프로토콜 수준입니다.

연습 문제

등록은 성공했지만 실행기 제출이 실패하는 상황을 구현해 봅니다. 결과는 STARTED, SERVER_CLOSING, EXECUTOR_REJECTED 세 가지로 구분하고, 거부된 세션은 목록에서 사라지며 close 횟수가 정확히 한 번이어야 합니다.

보상 순서를 포함한 구현

등록 실패에는 실행기 호출을 하지 않습니다. 제출 예외를 잡은 경로에서는 레지스트리 제거를 먼저 시도하고 세션 close를 호출합니다. close 자체도 제거를 수행할 수 있으므로 레지스트리 remove는 멱등 연산이어야 합니다.

solution/SessionAdmissionSolution.java

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

public final class SessionAdmissionSolution {
    enum Result {
        STARTED,
        SERVER_CLOSING,
        EXECUTOR_REJECTED
    }

    static final class Session implements Runnable, AutoCloseable {
        private final String id;
        private final AtomicInteger closeCount = new AtomicInteger();

        Session(String id) {
            this.id = id;
        }

        @Override
        public void run() {
            System.out.println("run=" + id);
        }

        @Override
        public void close() {
            closeCount.compareAndSet(0, 1);
        }
    }

    private final Set<Session> sessions = new HashSet<>();
    private final Executor executor;
    private boolean closing;

    SessionAdmissionSolution(Executor executor) {
        this.executor = executor;
    }

    synchronized void beginClose() {
        closing = true;
    }

    Result admit(Session session) {
        synchronized (this) {
            if (closing) {
                session.close();
                return Result.SERVER_CLOSING;
            }
            sessions.add(session);
        }

        try {
            executor.execute(session);
            return Result.STARTED;
        } catch (RejectedExecutionException error) {
            synchronized (this) {
                sessions.remove(session);
            }
            session.close();
            return Result.EXECUTOR_REJECTED;
        }
    }

    synchronized int activeCount() {
        return sessions.size();
    }

    public static void main(String[] args) {
        Executor rejecting = task -> {
            throw new RejectedExecutionException("stopped");
        };
        var admission = new SessionAdmissionSolution(rejecting);
        var session = new Session("s-9");

        System.out.println(admission.admit(session));
        System.out.println("active=" + admission.activeCount());
        System.out.println("closeCount=" + session.closeCount.get());
    }
}

성공 경로에서는 작업이 끝난 뒤 레지스트리에서 제거하는 책임이 Session.run()의 finally에 필요합니다. 연습 코드에서는 거부 보상에 집중하려고 정상 실행기의 제거 부분을 생략했으므로, 직접 onFinished 콜백을 추가해 수명 전체를 완성합니다. 그 뒤 즉시 실행기와 비동기 실행기 두 종류로 같은 불변식을 확인합니다.

세션 경계 검증

완료 판정은 목록 크기 하나로 끝나지 않습니다. 참가자 한 명의 전송 실패 뒤에도 다른 참가자가 메시지를 받는지, 느린 기록기가 등록 모니터를 점유하지 않는지, 종료 전환 뒤 새 세션이 거절되는지, 제출 실패가 등록과 소켓을 모두 보상하는지 확인합니다. 테스트가 끝난 뒤 활성 세션과 출력 작업자 수가 0이어야 합니다.

세션 레지스트리는 이후 사용자 이름 변경, /users 조회, 퇴장 알림의 기반이 됩니다. 다음 단계에서는 문자열 조건문에 명령이 계속 붙는 구조를 분해하고, 파싱 오류와 상태 위반을 연결 종료와 분리하는 명령 처리 계층을 설계합니다.