ネットワークアプリケーションの開発において、ノード間を移動する生データ（バイト列）と、アプリケーションが処理可能な構造化されたデータとの相互変換は不可欠です。この責任を負うのがコデック（Coder）であり、エンコーダーとデコーダーで構成されます。エンコーダーはメッセージを転送に適した形式（主にバイトストリーム）へ変換し、デコーダーはその逆のプロセスを担当します。具体的には、エンコーダーは送信側のデータを扱い、デコーダーは受信側のデータフローを処理します。
ByteToMessageDecoder の内部動作原理
ビジネスロジック層ではオブジェクトや文字列などの型が必要ですが、ネットワーク層からはバイトデータとして到着します。これを業務データに変換するための基本クラスが Netty の ByteToMessageDecoder です。
このクラスは ChannelInboundHandlerAdapter を継承しており、核心的な処理は channelRead メソッドに集約されます。以下のコードは、データ蓄積と変換の簡略化されたロジックを示しています。
// バッファ蓄積用フィールド
private ByteBuf pendingBuffer = null;
// デコード結果の出力リスト
private final List<Object> decodedResults = new ArrayList<>();
@Override
public void channelRead(ChannelHandlerContext ctx, Object obj) {
if (!(obj instanceof ByteBuf)) {
ctx.fireChannelRead(obj);
return;
}
ByteBuf input = (ByteBuf) obj;
boolean isFirstRead = (pendingBuffer == null);
try {
// 新規データの取り込みまたは結合
if (isFirstRead) {
pendingBuffer = input.retainedSlice();
} else {
// 既存バッファと新着データをマージ
ByteBuf combined = mergeBuffers(pendingBuffer, input);
pendingBuffer.release();
pendingBuffer = combined;
}
// 具体的なデコード処理を実行
callDecode(ctx, pendingBuffer, decodedResults);
} finally {
// リソース解放とフラグ管理
if (pendingBuffer != null && !pendingBuffer.isReadable()) {
pendingBuffer.release();
pendingBuffer = null;
}
// 次へデータを伝搬
fireNextHandlers(ctx, decodedResults);
decodedResults.clear();
}
}
上記の処理フローは以下のステップに分けられます。
- 入力検証: メッセージが ByteBuf 型か確認し、異なる場合は次のハンドラーへ委譲。
- パッキング対策: ネットワーク遅延による粘包・拆分に対応するため、不完全なデータは一旦
pendingBufferに保持します。 - データ統合: 初回受け取り時はそのまま割り当て、二回目以降は既存データと結合します。
- 解析実行:
callDecodeを呼び出し、変換可能かを試みます。 - 伝搬: 成功したオブジェクト群を後続のハンドラーに渡します。
callDecode メソッドは、バッファ内に十分なデータがある限りループを継続し、サブクラスの実装した decode メソッドを通じてオブジェクト生成を試みます。もし解析に必要なバイト数が不足している場合、in.readableBytes() が変化しないことを検知してループを終了し、より多くのデータ到着を待ちます。
累積バッファのマージ戦略
複数のデータブロックを一つにまとめる処理は、Cumulator インターフェースによって制御されます。主要な判断基準は、現在のバッファ容量が新しいデータを格納できるかどうかです。
/**
* バッファ結合処理
*/
public ByteBuf mergeBuffers(ByteBufAllocator alloc, ByteBuf existing, ByteBuf incoming) {
// 容量チェック：書き込みインデックス + 新しいサイズ > 最大容量、または参照カウントが 1 超の場合
if (existing.writerIndex() >= existing.maxCapacity() - incoming.readableBytes() || existing.refCnt() > 1) {
// 容量拡張が必要な場合、新しいバッファを確保
ByteBuf expanded = expandBuffer(alloc, existing, incoming.readableBytes());
return appendData(expanded, incoming);
}
// そのまま使用可能ならそのまま返却
return appendData(existing, incoming);
}
private ByteBuf appendData(ByteBuf buffer, ByteBuf source) {
buffer.writeBytes(source);
source.release();
return buffer;
}
必要に応じてメモリ再割り配を行い、古いバッファを解放することで、メモリの断片化を防ぎつつ効率よくデータを蓄積します。
LengthFieldBasedFrameDecoder によるメッセージ境界の特定
特定の長さを持つデータフレームを抽出する場合、ByteToMessageDecoder のサブクラスである LengthFieldBasedFrameDecoder が有効です。これはヘッダー部分にペイロードの長さ情報を埋め込むプロトコル向けに設計されています。
コンストラクタでは、バイトオーダーや長さフィールドのオフセット位置などを設定します。例えば lengthFieldOffset=0 で lengthFieldLength=4 とすると、バッファの先頭 4 バイトからデータ長を取得し、その後に続くバイト数分を読み取る仕組みになります。
protected Object decode(ChannelHandlerContext ctx, ByteBuf input) throws Exception {
// 過大なフレームの破棄処理中の判定
if (isDroppingLongFrame) {
handleDrop(input);
return null;
}
// ヘッダー情報の読み込み可否確認
int readOffset = input.readerIndex();
if (input.readableBytes() < lengthFieldEndOffset) {
return null; // 不全なので保留
}
// 実際の長さフィールド位置の計算
int lengthPos = readOffset + lengthFieldOffset;
// バイト数に応じた値の取得
long payloadLen = extractLengthValue(input, lengthPos, lengthFieldLength, byteOrder);
// 負の長さチェックなどバリエーション調整
if (payloadLen < 0) {
throw new DecoderException("Invalid negative length");
}
// 補正値を加算した最終的なフレームサイズ
long adjustedFrameSize = payloadLen + lengthAdjustment + lengthFieldEndOffset;
// 最大許容サイズ超過の監視
if (adjustedFrameSize > maxFrameLength) {
exceedLimit(input, adjustedFrameSize);
return null;
}
// 現在読み込めるデータ量が十分か確認
int requiredBytes = (int) adjustedFrameSize;
if (input.readableBytes() < requiredBytes) {
return null; // データ不足のため待機
}
// スキップすべきバイト分の移動
input.skipBytes(initialBytesToStrip);
// 完全なフレームの抽出とインデックス更新
int startIndex = input.readerIndex();
int sliceLen = requiredBytes - initialBytesToStrip;
ByteBuf extractedFrame = input.slice(startIndex, sliceLen);
// 読込ポインタを次に進める
input.readerIndex(startIndex + sliceLen);
return extractedFrame;
}
キーメソッドの詳細
extractLengthValue は、指定されたバイト数（1, 2, 4, 8 など）に基づいて整数値を復元します。一方、slice 操作においては retainedSlice を利用することが多く、これは元のバッファのインデックスを変更せずに、独立した参照を持つ新たな ByteBuf オブジェクトを生成するために重要です。これにより、元のデータのライフサイクルに影響を与えずにフレームを抽出できます。
全体として、このデコーダーは以下の方針で動作します。
- バッファ上の指定位置から長さ値をリード。
- 長さ値と利用可能なバイト数の比較を行い、パケット不完整の場合は中断。
- 不要なヘッダー部分をスキップ。
- 定義された長さ分のデータを切り出して次の処理へ渡し、読み込みポインタを進める。