채팅 클라이언트 입출력

채팅 클라이언트는 요청을 보내고 응답 하나를 받는 단순한 프로그램이 아닙니다. 사용자가 아무것도 입력하지 않는 순간에도 다른 참가자의 메시지는 도착할 수 있고, 서버가 연결을 종료하는 동안 사용자는 콘솔 입력에서 멈춰 있을 수 있습니다. 따라서 한 스레드가 nextLine()과 readUTF()를 번갈아 호출하는 구조는 양방향 통신 요구와 맞지 않습니다. 두 입력원은 서로 독립적으로 진행해야 합니다.

설계를 시작할 때 화면 출력, 콘솔 입력, 소켓 읽기, 소켓 쓰기를 한 덩어리로 묶지 않습니다. 클라이언트가 지켜야 할 핵심 계약은 세 가집니다. 로컬 입력이 지연되어도 원격 메시지를 읽을 수 있어야 하고, 사용자가 적은 평문을 프로토콜 명령으로 변환해야 하며, 어느 한쪽에서 종료가 시작되면 모든 블로킹 지점을 깨우고 자원을 한 번만 닫아야 합니다. 이 계약을 먼저 고정하면 GUI나 파일 입력으로 바뀌어도 통신 코어는 유지됩니다.

입력 역할 충돌

다음 코드는 컴파일되지만 실시간 채팅 클라이언트로는 동작하지 않습니다. 사용자가 엔터를 누르기 전까지 서버 메시지를 읽는 코드에 도달하지 못합니다. 반대로 readUTF()를 먼저 호출하면 서버가 보낼 데이터가 없을 때 콘솔 입력을 처리하지 못합니다. 호출 순서를 뒤집는 것으로는 문제가 해결되지 않습니다.

bad/AlternatingChatClient.java

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.util.Scanner;

public final class AlternatingChatClient {
    static void run(
            Scanner console,
            DataInputStream input,
            DataOutputStream output) throws Exception {
        while (true) {
            String typed = console.nextLine();
            output.writeUTF(typed);

            String received = input.readUTF();
            System.out.println(received);
        }
    }

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

이 구조는 “내가 한 번 보내야 상대 메시지를 한 번 읽는다”라는 잘못된 순서를 프로토콜에 강제합니다. 서버가 공지 두 개를 연속으로 보내거나 다른 사용자가 먼저 말하면 수신 데이터가 소켓 버퍼에 쌓입니다. 더 심각한 경우에는 양쪽이 모두 상대의 입력을 기다려 진행이 멈춥니다. 네트워크 프로그램을 읽을 때는 각 블로킹 호출이 어떤 사건으로 깨어나는지, 그 사건을 다른 역할이 만들어낼 수 있는지를 표시해야 합니다.

읽기 역할은 서버 프레임만 소비하고, 쓰기 역할은 로컬 입력을 프레임으로 바꾸어 전송합니다. 두 역할은 같은 연결을 공유하지만 한 스트림을 여러 스레드가 동시에 읽거나 쓰게 하지 않습니다. DataInputStream은 수신자 하나, DataOutputStream은 송신자 하나가 소유하도록 제한하면 프레임 경계가 섞이는 문제도 피할 수 있습니다.

명령 프레임 경계

서버가 /join|Mina, /message|hello, /users, /exit 형식을 처리한다고 해도 UI 계층 전체가 문자열 결합 규칙을 알아서는 안 됩니다. 명령 타입과 코덱을 두면 잘못된 구분자, 빈 이름, 알 수 없는 명령을 소켓에 쓰기 전에 거절할 수 있습니다. split("\\|", 2)의 두 번째 인수는 메시지 본문에 구분자가 더 있어도 첫 경계만 나누기 위한 선택입니다.

src/ChatCommandCodec.java

import java.util.Objects;

public final class ChatCommandCodec {
    sealed interface Command
            permits Join, Message, ChangeName, ListUsers, Exit {
    }

    record Join(String name) implements Command {
        Join {
            requireText(name, "name");
        }
    }

    record Message(String text) implements Command {
        Message {
            requireText(text, "message");
        }
    }

    record ChangeName(String name) implements Command {
        ChangeName {
            requireText(name, "name");
        }
    }

    record ListUsers() implements Command {
    }

    record Exit() implements Command {
    }

    static String encode(Command command) {
        return switch (command) {
            case Join join -> "/join|" + join.name();
            case Message message -> "/message|" + message.text();
            case ChangeName change -> "/change|" + change.name();
            case ListUsers ignored -> "/users";
            case Exit ignored -> "/exit";
        };
    }

    static Command decode(String frame) {
        Objects.requireNonNull(frame, "frame");
        String[] parts = frame.split("\\|", 2);
        return switch (parts[0]) {
            case "/join" -> new Join(argument(parts, frame));
            case "/message" -> new Message(argument(parts, frame));
            case "/change" -> new ChangeName(argument(parts, frame));
            case "/users" -> new ListUsers();
            case "/exit" -> new Exit();
            default -> throw new IllegalArgumentException("unknown frame: " + frame);
        };
    }

    private static String argument(String[] parts, String frame) {
        if (parts.length != 2) {
            throw new IllegalArgumentException("argument missing: " + frame);
        }
        return parts[1];
    }

    private static void requireText(String value, String field) {
        if (value == null || value.isBlank()) {
            throw new IllegalArgumentException(field + " must not be blank");
        }
    }

    public static void main(String[] args) {
        Command command = decode("/message|hello|team");
        System.out.println(command);
        System.out.println(encode(new Join("Mina")));
        System.out.println(encode(command));
    }
}

코덱은 전송 형식만 책임지고, “입장하기 전에 메시지를 보낼 수 있는가” 같은 세션 규칙은 상태 계층이 판단합니다. 문자열 파싱과 업무 순서를 한 메서드에 합치면 문법 오류와 상태 오류가 같은 예외로 나타나 원인을 분류하기 어렵습니다. 또 메시지 길이 상한을 코덱 입구에 두면 매우 큰 입력이 서버까지 전달되어 메모리를 압박하는 것을 막을 수 있습니다.

명령 프레임은 writeUTF()가 제공하는 길이 경계 안에서만 유효합니다. 이 메서드는 수정 UTF 형식과 최대 길이 제약을 가지므로 범용 채팅 프로토콜로 확장할 때는 명시적인 UTF-8 길이 접두사나 줄 기반 프레임을 검토해야 합니다. 여기서는 학습 범위를 좁히기 위해 양쪽 모두 같은 Java 데이터 스트림을 사용한다는 조건을 문서화합니다.

메모리 채널 검증

실제 소켓과 콘솔만으로 테스트하면 입력 타이밍에 따라 결과가 달라집니다. 먼저 BlockingQueue를 네트워크와 콘솔의 대역으로 사용해 읽기와 쓰기가 독립적으로 진행되는지 확인합니다. 아래 예제에서 로컬 작성자는 두 명령을 서버 방향 큐에 넣고, 원격 수신자는 사용자가 더 입력하지 않아도 서버 공지를 화면 목록에 추가합니다.

src/DuplexClientHarness.java

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public final class DuplexClientHarness {
    private static final String CLOSED = "__closed__";

    static List<String> exchange(
            List<String> localLines,
            List<String> serverFrames) throws Exception {
        BlockingQueue<String> outbound = new LinkedBlockingQueue<>();
        BlockingQueue<String> inbound = new LinkedBlockingQueue<>();
        List<String> displayed = new ArrayList<>();

        try (var tasks = java.util.concurrent.Executors
                .newVirtualThreadPerTaskExecutor()) {
            var writer = tasks.submit(() -> {
                for (String line : localLines) {
                    String frame = line.startsWith("/")
                            ? line
                            : "/message|" + line;
                    outbound.put(frame);
                }
                outbound.put(CLOSED);
                return null;
            });

            var reader = tasks.submit(() -> {
                while (true) {
                    String frame = inbound.take();
                    if (frame.equals(CLOSED)) {
                        return null;
                    }
                    synchronized (displayed) {
                        displayed.add(frame);
                    }
                }
            });

            for (String frame : serverFrames) {
                inbound.put(frame);
            }
            inbound.put(CLOSED);

            writer.get(1, TimeUnit.SECONDS);
            reader.get(1, TimeUnit.SECONDS);

            List<String> sent = new ArrayList<>();
            String frame;
            while ((frame = outbound.poll(10, TimeUnit.MILLISECONDS)) != null) {
                if (!frame.equals(CLOSED)) {
                    sent.add(frame);
                }
            }
            System.out.println("sent=" + sent);
            return List.copyOf(displayed);
        }
    }

    public static void main(String[] args) throws Exception {
        List<String> displayed = exchange(
                List.of("hello", "/users"),
                List.of("Mina joined", "[Mina] welcome"));
        System.out.println("displayed=" + displayed);
    }
}

이 실험은 출력 순서를 무리하게 고정하지 않습니다. 중요한 관측값은 송신 프레임 두 개가 정확히 만들어지고, 수신 프레임 두 개가 입력 대기와 무관하게 처리되었다는 사실입니다. 운영 코드에서는 각 역할의 스레드 이름, 마지막 성공 프레임 시각, 종료 원인을 기록하면 멈춤 위치를 찾기 쉽습니다. 큐 크기를 무제한으로 두는 것은 테스트 편의를 위한 것이며 실제 UI 이벤트 큐에는 상한과 드롭 정책이 필요합니다.

클라이언트 종료 전파

읽기 역할이 EOF나 IOException을 만나면 쓰기 역할이 콘솔에서 계속 기다릴 수 있습니다. 반대로 사용자가 /exit를 입력해도 수신 역할은 서버가 연결을 닫기 전까지 read에서 멈춥니다. 그래서 두 작업 각각이 자기 자원만 닫는 구조보다, 클라이언트 수명 객체에 종료 요청을 위임하는 편이 낫습니다. 수명 객체는 첫 종료 원인을 저장하고 소켓을 닫아 네트워크 블로킹을 해제하며 UI 입력원에는 별도 취소 신호를 보냅니다.

System.in.close()는 예제에서는 콘솔 대기를 깨울 수 있지만 프로세스 전역 자원을 닫는 강한 동작입니다. 같은 JVM에서 다른 기능도 표준 입력을 사용한다면 예상치 못한 영향을 줍니다. 실제 애플리케이션은 입력 어댑터를 별도 스레드에서 소유하고, 중단 가능한 UI 이벤트나 종료 토큰을 제공하는 것이 안전합니다. 닫을 수 없는 블로킹 API를 선택했다면 데몬 스레드 여부와 프로세스 종료 전략까지 명시해야 합니다.

종료 사건	먼저 수행할 일	상대 역할을 깨우는 수단	사용자에게 보일 결과
사용자의 `/exit`	프레임 전송을 한 번 시도	서버 응답 뒤 소켓 `close`	정상 퇴장
서버 EOF	종료 원인을 `REMOTE_EOF`로 기록	로컬 입력 어댑터 취소	서버가 연결 종료
읽기 오류	소켓을 멱등하게 `close`	쓰기 작업 취소	연결 오류
UI 닫기	새 입력 수락 중지	소켓과 작업 실행기 종료	사용자가 취소

닫기 호출은 여러 경로에서 경쟁하므로 AtomicBoolean.compareAndSet(false, true)처럼 단일 전환을 사용합니다. 단, 플래그만 바꾸고 블로킹 자원을 실제로 닫지 않으면 작업은 종료되지 않습니다. 상태 전환과 해제 동작을 같은 메서드에 모으고, 두 번째 호출은 조용히 반환하도록 만듭니다.

연습 문제

다음 요구를 만족하는 ClientLifetime을 작성해 봅니다.

close(reason)은 여러 스레드에서 동시에 호출할 수 있습니다.
가장 먼저 전달된 이유만 결과로 남깁니다.
등록된 두 취소 작업은 각각 정확히 한 번 실행합니다.
취소 작업 하나가 실패해도 나머지 작업은 실행합니다.

해설과 한 가지 구현

첫 이유와 닫힘 여부를 따로 관리하면 두 값이 엇갈릴 수 있습니다. 하나의 불변 State를 AtomicReference로 교체하면 전환과 이유 저장이 같은 원자 연산에 묶입니다. 자원 해제 오류는 목록으로 모아 호출자에게 돌려줍니다.

solution/ClientLifetimeSolution.java

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

public final class ClientLifetimeSolution {
    enum Reason {
        USER_EXIT,
        REMOTE_EOF,
        READ_FAILURE,
        UI_CLOSED
    }

    record State(boolean closed, Reason reason) {
        static State open() {
            return new State(false, null);
        }
    }

    private static final State OPEN = State.open();

    private final AtomicReference<State> state =
            new AtomicReference<>(OPEN);
    private final List<Runnable> cancelActions;

    ClientLifetimeSolution(List<Runnable> cancelActions) {
        this.cancelActions = List.copyOf(cancelActions);
    }

    List<RuntimeException> close(Reason reason) {
        State closed = new State(true, reason);
        if (!state.compareAndSet(OPEN, closed)) {
            return List.of();
        }

        List<RuntimeException> failures = new ArrayList<>();
        for (Runnable action : cancelActions) {
            try {
                action.run();
            } catch (RuntimeException error) {
                failures.add(error);
            }
        }
        return List.copyOf(failures);
    }

    State snapshot() {
        return state.get();
    }

    public static void main(String[] args) {
        List<String> events = new ArrayList<>();
        var lifetime = new ClientLifetimeSolution(List.of(
                () -> events.add("socket closed"),
                () -> events.add("input cancelled")));

        System.out.println(lifetime.close(Reason.REMOTE_EOF));
        System.out.println(lifetime.close(Reason.USER_EXIT));
        System.out.println(lifetime.snapshot());
        System.out.println(events);
    }
}

OPEN을 단일 상수로 보관한 이유는 CAS가 record의 값 동등성이 아니라 참조 동일성을 비교하기 때문입니다. 연습을 확장하려면 동시에 열 개 스레드가 서로 다른 이유로 close를 호출하게 만들고, 취소 로그가 두 줄만 남으며 결과 이유가 호출자 중 하나인지 검사합니다.

완성 기준

자습 단계에서 화면에 메시지가 보였다는 결과만으로 완료라고 판단하지 않습니다. 사용자가 입력하지 않아도 서버 공지가 출력되는지, 프레임 문법 오류가 전송 전에 구분되는지, EOF와 사용자 종료가 동일한 close 경계로 모이는지 각각 독립적으로 확인합니다. 마지막으로 닫힌 뒤 남은 비데몬 스레드가 없는지 스레드 덤프나 실행기 종료 상태로 검증해야 합니다.

이 구조의 핵심은 스레드를 두 개 만들었다는 사실이 아니라 역할별 블로킹과 소유권을 드러낸 데 있습니다. 다음 문서에서는 서버가 연결별 세션을 등록하고, 한 사용자의 메시지를 다른 모든 세션으로 전달할 때 잠금과 느린 수신자를 어떻게 다루는지 이어서 살펴봅니다.