실시간 STT voice engine 개선사항
기존 AsyncSocketChannel 사용 부분을 WebFlux Netty 기반 NIO 로 변경
뭐가다름?
AsynchronousSocketChannel
JDK NIO2 기반
로우 레벨 TCP 소켓
Completion Handler 기반
연결, 읽기, 쓰기, 버퍼 처리, 메시지 경계 처리 등을 직접 구현해야 함
대표 기능
- 원격 서버에 비동기 connect
- 비동기 read
- 비동기 write
- CompletionHandler 또는 Future 기반 결과 처리
일반적인 흐름
- 채널 연다
- 서버에 connect 한다
- 연결되면 write 한다
- 응답을 read 한다
- 필요하면 다시 write/read 반복
- 끝나면 close 한다
메시지 단위가 Byte stream
서버가 'HELLO' 라는 메시지를 보내도
실제 수신은
'HEL'
'LO'
혹은
'HELL'
'O'
와 같이 read() 한번에 메시지 하나 보장 되지 않음
** voice engine에서는 고정 길이 기반으로 구성
- 이전 read에서 남은 바이트 보관
- 새로 읽은 데이터 누적
- 완전한 메시지 하나가 되었는지 판단
- 완전한 메시지만 꺼내서 처리
와 같이 처리
또한
ByteBuffer를 직접 컨트롤
예를 들어
- clear()
- flip()
- compact()
- remaining()
흐름을 매우 엄격하게 준수
read() 후 flip() 하지 않으면 읽을 수 없고
clear() 해버리면 메시지 증발도 가능
또한
heartbeat 직접해야됨 ㅠ
example
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
public class AsyncTcpClientExample {
public static void main(String[] args) throws Exception {
// 비동기 TCP 채널 open
AsynchronousSocketChannel client = AsynchronousSocketChannel.open();
InetSocketAddress serverAddress = new InetSocketAddress("127.0.0.1", 9000);
// 서버에 비동기 연결 시도
// 성공 : completed
// 실패 : failed
client.connect(serverAddress, null, new CompletionHandler<Void, Void>() {
@Override
public void completed(Void result, Void attachment) {
System.out.println("서버 연결 성공");
String message = "hello server\n";
// 문자열을 byte로 바꿔 서버에 전송
// CompletionHandler의 Integer 는 실제로 쓴 바이트
ByteBuffer writeBuffer = ByteBuffer.wrap(message.getBytes(StandardCharsets.UTF_8));
client.write(writeBuffer, null, new CompletionHandler<Integer, Void>() {
@Override
public void completed(Integer bytesWritten, Void attachment) {
System.out.println("전송 완료 bytes = " + bytesWritten);
// 서버 응답을 읽음
// byteRead == -1 이면 상대가 연결 해제
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
client.read(readBuffer, null, new CompletionHandler<Integer, Void>() {
@Override
public void completed(Integer bytesRead, Void attachment) {
if (bytesRead == -1) {
System.out.println("서버가 연결 종료");
closeClient(client);
return;
}
// 소켓이 버퍼에 데이터를 쓴 뒤 flip해줘야 읽기 가능
readBuffer.flip();
byte[] data = new byte[readBuffer.remaining()];
readBuffer.get(data);
String response = new String(data, StandardCharsets.UTF_8);
System.out.println("응답 수신 = " + response);
closeClient(client);
}
@Override
public void failed(Throwable exc, Void attachment) {
System.out.println("읽기 실패: " + exc.getMessage());
closeClient(client);
}
});
}
@Override
public void failed(Throwable exc, Void attachment) {
System.out.println("쓰기 실패: " + exc.getMessage());
closeClient(client);
}
});
}
@Override
public void failed(Throwable exc, Void attachment) {
System.out.println("연결 실패: " + exc.getMessage());
closeClient(client);
}
});
// 메인 스레드가 너무 빨리 종료되면 비동기 콜백 실행 전에 프로그램이 종료될 수 있음
// 실제 서버에서는 프로세스가 살아있으니 이런식으로 구성 안함
Thread.sleep(5000);
}
private static void closeClient(AsynchronousSocketChannel client) {
try {
if (client != null && client.isOpen()) {
client.close();
}
} catch (Exception e) {
System.out.println("소켓 종료 중 오류: " + e.getMessage());
}
}
}
** CompletionHandler ??
비동기 작업 완료 시 호출되는 콜백 인터페이스
기본 구조
new CompletionHandler<ResultType, AttachmentType>() {
@Override
public void completed(ResultType result, AttachmentType attachment) {
}
@Override
public void failed(Throwable exc, AttachmentType attachment) {
}
}
- connect 결과 타입: Void
- read/write 결과 타입: Integer
** Attachment ??
connect, read, write 호출 시 넘기는 두번째 인자
콜백에서 사용할 데이터를 넘겨줄 수 있음
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
ByteBuffer buffer = ByteBuffer.allocate(1024);
channel.read(buffer, "첫 번째 read 작업", new CompletionHandler<Integer, String>() {
@Override
public void completed(Integer result, String attachment) {
System.out.println("완료된 작업: " + attachment);
System.out.println("읽은 바이트 수: " + result);
}
@Override
public void failed(Throwable exc, String attachment) {
System.out.println("실패한 작업: " + attachment);
}
});
이런식으로 '첫번째 read작업' 이라는 문자열을 completed, failed 콜백에서 사용 가능
실무에서는 보통 상태 객체로 넘김
좀 더 나아가서
재귀형 read 루프
private static void readLoop(AsynchronousSocketChannel client, ByteBuffer buffer) {
buffer.clear();
client.read(buffer, null, new CompletionHandler<Integer, Void>() {
@Override
public void completed(Integer bytesRead, Void attachment) {
if (bytesRead == -1) {
System.out.println("연결 종료");
closeClient(client);
return;
}
buffer.flip();
byte[] data = new byte[buffer.remaining()];
buffer.get(data);
System.out.println("수신: " + new String(data, StandardCharsets.UTF_8));
// complted 콜백 내에서 readLoop 메서드 재호출
// 실제로는 전달 받은 데이터가 완전한지 판단-처리 로직 필요
readLoop(client, buffer);
}
@Override
public void failed(Throwable exc, Void attachment) {
System.out.println("읽기 실패: " + exc.getMessage());
closeClient(client);
}
});
}
Netty
비동기 TCP를 만들기 위한 프레임워크
이벤트 루프, 채널 생명주기, 파이프라인, 디코더/인코더 와 같은 구조를 제공
네트워크 프로그램을 체계적으로 작성하게 해줌
>> TCP 비동기 처리를 더 큰 구조로 감싼 프레임워크
쉽게 말해
- 연결이 열렸다
- 데이터가 들어왔다
- 쓰기가 끝났다
- 예외가 났다
- 연결이 닫혔다
이런 이벤트를 루프가 받아서 처리
>> 네트워크 흐름이 일관적
핵심 구성요소
- EventLoop
이벤트 루프
이벤트를 처리하는 루프
TCP 연결에서 발생하는
- 연결됨
- 데이터 도착
- 쓰기 완료
- 연결 종료
- 예외 발생
위와 같은 이벤트를 이벤트 루프가 처리
하나의 channel 은 하나의 EventLoop에 등록
channel.eventLoop() 로 확인 가능
- Channel
연결 자체를 표현하는 객체
TCP 입장에서 클라이언트 하나가 붙으면 그 연결 하나가 하나의 Channel 이라고 봄
이 채널을 통해
- 원격 주소 확인
- 데이터 write
- close
- pipeline 접근
- eventLoop 접근
Channel.closeFuture() : 채널 닫힐 때 완료되는 future
flush() : pending 메시지를 실제로 flush 하는 동작
즉, 소켓 연결과 관련된 행동을 channel 이 대표로 함
- ChannelPipeline
Netty의 가장 중요한 개념
- raw bytes 수신
- 프레임 분리
- 문자열 디코딩
- 비즈니스 처리
- 응답 문자열 생성
- 바이트 인코딩
- 전송
한 덩어리보다는 단계별 처리 가능
- Handler
- ChannelInboundHandlerAdapter
- SimpleChannelInboundHandler<T>
와 같은 클래스를 사용하여 이벤트 처리 가능
- ByteBuf
ByteBuffer 대신 ByteBuf 사용
예제 코드
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import java.nio.charset.StandardCharsets;
public class SimpleNettyTcpServer {
public static void main(String[] args) throws Exception {
int port = 9000;
// Netty에서는 보통 두 그룹을 띄움
// bossGroup : 새 연결 accept
// workerGroup : 실제 읽기 처리
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
// 서버 띄우기 위한 설정 객체
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
// NIO 기반 소켓 채널 사용
.channel(NioServerSocketChannel.class)
// 새 클라이언트가 연결 될 때 마다 연결용 pipeline 초기화
// 클라이언트 A가 붙으면 A용 파이프라인, B면 B용 파이프라인..
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new ChannelInboundHandlerAdapter() {
// 연결 직후 호출되는 함수
// WebSocket의 onOpen 같은
@Override
public void channelActive(ChannelHandlerContext ctx) {
System.out.println("클라이언트 연결됨: " + ctx.channel().remoteAddress());
}
// 클라이언트가 보낸 데이터 수신 시 호출
// onMessage 같은
// 여기서 msg는 기본적으로 ByteBuf
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
// 수신한 ByteBuf를 문자열로 변환하는 단순 예제
// 실무에서는 TCP는 stream이기 때문에 메시지 경계가 모호
// Netty에서는 ByteToMessageDecoder와 각종 frameDecoder로
// fragmentation/reassembly 문제 해결
ByteBuf buf = (ByteBuf) msg;
try {
String received = buf.toString(StandardCharsets.UTF_8);
System.out.println("수신: " + received);
String response = "echo: " + received;
// ByteBuf를 만들고 writeAndFlush()로 바로 전송
// write : outbound 큐에 넣음
// flush : 실제로 흘려 보냄
// writeAndFlush : 두개 동시에
ByteBuf out = ctx.alloc().buffer();
out.writeCharSequence(response, StandardCharsets.UTF_8);
ctx.writeAndFlush(out);
} finally {
// Netty의 ByteBuf는 reference-counted 객체이기 때문에
// 직접 수신하면 직접 realease 필요
buf.release();
}
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
System.out.println("클라이언트 연결 종료: " + ctx.channel().remoteAddress());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
});
}
})
.childOption(ChannelOption.SO_KEEPALIVE, true);
ChannelFuture bindFuture = bootstrap.bind(port).sync();
System.out.println("서버 시작 완료. port = " + port);
bindFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
위는 예시이기 때문에 많은 한계 존재
HELLO 를 보내면 서버는
H
ELL
O
이렇게 수신 할 수 도 있음
** 실무에선 프레이밍 필요!
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import java.nio.charset.StandardCharsets;
public class LineBasedNettyServer {
public static void main(String[] args) throws Exception {
int port = 9000;
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline p = ch.pipeline();
p.addLast(new DelimiterBasedFrameDecoder(1024, Delimiters.lineDelimiter()));
p.addLast(new StringDecoder(StandardCharsets.UTF_8));
p.addLast(new StringEncoder(StandardCharsets.UTF_8));
p.addLast(new SimpleChannelInboundHandler<String>() {
@Override
public void channelActive(ChannelHandlerContext ctx) {
System.out.println("연결됨: " + ctx.channel().remoteAddress());
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) {
System.out.println("한 줄 수신: " + msg);
ctx.writeAndFlush("echo: " + msg + "\n");
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
});
}
});
ChannelFuture future = bootstrap.bind(port).sync();
System.out.println("라인 기반 서버 시작: " + port);
future.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
이 예제에서 pipeline
- DelimiterBasedFrameDecoder
- StringDecoder
- StringEncoder
- 비즈니스 핸들러
즉 데이터가 들어오면:
- raw bytes
- 줄 단위 frame 분리
- String 변환
- 네 비즈니스 처리
순서로 동작
-- AsyncronousSocketChannel 에서 직접 해야하던
- 누적 버퍼 유지
- \n 찾기
- 남은 데이터 보관
- 메시지 잘라내기
를 프레임 디코더가 대신해주는 구조
