博客

概述

编码器操作出站数据，而解码器处理入站数据。编码器是将消息转换为适合于传输的格式，而对应的解码器则是将网络字节流转换回应用程序的消息格式。

解码器

• 将字节解码为消息：ByteToMessageDecoder、ReplayingDecoder
• 将一种消息类型解码为另外一种：MessageToMessageDecoder

因为解码器是负责将入站数据从一种格式转换到另外一种格式，

所以Netty的解码器实现了ChannelInboundHandler

。每当需要为ChannelPipeline中的下一个ChannelInboundHandler转换入站数据是会用到，

此外可以将多个解码器连接在一起

，以实现复杂的转换逻辑。

// 扩展ByteToMessageDecoder类，以将字节解码为特定格式
public class ToIntegerDecoder extends ByteToMessageDecoder{
    @override
    public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception{
        // 检查是否至少有4个字节可读
        if (in.readableBytes() >= 4){
            // 从入站ByteBuf中读取一个int，并将其添加到解码消息的List中
            out.add(in.readInt());
        }
    }
}

编码器

• MessageToByteEncoder
• MessageToMessageEncoder
  
  +encode(ChannelHandlerContext ctx, I msg, ByteBuf out)

  encode()方法是你需要实现的唯一抽象方法。它被调用时 将会传入要被该类编码为 ByteBuf 的（类型为 I 的）出站消息 。该 ByteBuf 随后 将会被转发给 ChannelPipeline中的下一个 ChannelOutboundHandler

public class ShortToByteEncoder extends MessageToByteEncoder<Short>{
    @override
    public void encode(ChannelHandlerContext ctx, Short msg, ByteBUf out) throws Exception{
        // 将Short写入ByteBuf中
        out.writeShort(msg);
    }
}

预置的ChannelHandler和编解码器

1. Http编解码器

Http

请求

的组成部分

Http

响应

的组成部分

一个 HTTP 请求/响应可能由多个数据部分组成，并且它总是以一个 LastHttpContent 部分作为结束。

FullHttpRequest 和 FullHttpResponse 消息是特殊的子类型，分别代表了完整的请求和响应

。所有类型的 HTTP 消息（ FullHttpRequest、

LastHttpContent 以及代码清单 11-2 中展示的那些）都实现了 HttpObject 接口。

import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.*;

public class HttpPiplineInitialize extends ChannelInitializer<Channel> {


    private final boolean client;
    public HttpPiplineInitialize(boolean client){
        this.client = client;
    }
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        if (client){
            // 如果是客户端，则添加HttpResponseDecoder 以处理来自服务器的响应
            pipeline.addLast("decoder", new HttpResponseDecoder());
            // 如果是客户端，则添加 HttpRequestEncoder 以向服务器发送请求
            pipeline.addLast("encoder", new HttpRequestEncoder());


            // 一种聚合器写法：
            /*pipeline.addLast("codec", new HttpClientCodec());*/
        } else {
            // 如果是服务器，则添加 HttpResponseEncoder 以向客户端发送响应
            pipeline.addLast("decoder", new HttpRequestDecoder());
            // 如果是服务器，则添加 HttpRequestDecoder 以接收来自客户端的请求
            pipeline.addLast("encoder", new HttpResponseEncoder());

            // 一种聚合器写法
           /* pipeline.addLast("codec", new HttpServerCodec());*/
        }
    }
}

2. WebSocket

WebSocket解决了一个长期存在的问题：既然底层的协议（ HTTP） 是一个请求/响应模式的交互序列，

那么如何实时地发布信息呢？

AJAX提供了一定程度上的改善，但是数据流仍然是由客户端所发送的请求驱动的。

WebSocket提供了“在一个


单个的TCP连接上提供双向的通信


……结合WebSocket API……



它为网页和远程服务器之间的双向通信提供了一种替代HTTP轮询的方案



。”WebSocket 现在可以用于传输任意类型的数据， 很像普通的套接字。

展示了一个使用WebSocketServerProtocolHandler的简单示例，这个类处理协议升级握手，以及

3 种控制帧——Close、 Ping和Pong

。

Text和Binary数据帧将会被传递给下一个（由你实现的） ChannelHandler进行处理。

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;

public class WebSocketServerInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel channel) throws Exception {
        channel.pipeline().addLast(
                // 为握手提供聚合的HttpRequest
                new HttpServerCodec(),
                new HttpObjectAggregator(65535),
                // 如果被请求的端点是“/webSocket"则处理该升级握手
                new WebSocketServerProtocolHandler("/websocket"),
                // 处理TextWebSocketFrame
                new TextFrameHandler(),
                // 处理BinaryWebSocketFrame
                new BinaryFrameHandler(),
                // 处理ContinuationWebSocketFrame
                new ContinuationFrameHandler()
        );
    }

    public static final class TextFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame>{
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {
            // 处理TextWebSocketFrame
        }
    }

    public static final class BinaryFrameHandler extends SimpleChannelInboundHandler<BinaryWebSocketFrame>{
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, BinaryWebSocketFrame textWebSocketFrame) throws Exception {
            // 处理BinaryWebSocketFrame
        }
    }

    public static final class ContinuationFrameHandler extends SimpleChannelInboundHandler<ContinuationWebSocketFrame>{
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, ContinuationWebSocketFrame textWebSocketFrame) throws Exception {
            // 处理ContinuationWebSocketFrame
        }
    }
}