• Netty 编码器和解码器
    • 实现 Memcached 编码器
    • 实现 Memcached 解码器

    Netty 编码器和解码器

    Netty 的是一个复杂和先进的框架,但它并不玄幻。当我们请求一些设置了 key 的给定值时,我们知道 Request 类的一个实例被创建来代表这个请求。但 Netty 并不知道 Request 对象是如何转成 Memcached 所期望的。Memcached 所期望的是字节序列；忽略使用的协议，数据在网络上传输永远是字节序列。

    将 Request 对象转为 Memcached 所需的字节序列，Netty 需要用 MemcachedRequest 来编码成另外一种格式。这里所说的另外一种格式不单单是从对象转为字节，也可以是从对象转为对象，或者是从对象转为字符串等。编码器的内容可以详见第七章。

    Netty 提供了一个抽象类称为 MessageToByteEncoder。它提供了一个抽象方法,将一条消息(在本例中我们 MemcachedRequest 对象)转为字节。你显示什么信息实现通过使用 Java 泛型可以处理;例如 , MessageToByteEncoder 说这个编码器要编码的对象类型是 MemcachedRequest

    MessageToByteEncoder 和 Java 泛型

    使用 MessageToByteEncoder 可以绑定特定的参数类型。如果你有多个不同的消息类型，在相同的编码器里，也可以使用MessageToByteEncoder，注意检查消息的类型即可

    这也适用于解码器,除了解码器将一系列字节转换回一个对象。
    这个 Netty 的提供了 ByteToMessageDecoder 类,而不是提供一个编码方法用来实现解码。在接下来的两个部分你看看如何实现一个 Memcached 解码器和编码器。在你做之前,应该意识到在使用 Netty 时，你不总是需要自己提供编码器和解码器。自所以现在这么做是因为 Netty 没有对 Memcached 内置支持。而 HTTP 以及其他标准的协议，Netty 已经是提供的了。

    编码器和解码器

    记住,编码器处理出站，而解码器处理入站。这基本上意味着编码器将编码数据,写入远端。解码器将从远端读取处理数据。重要的是要记住,出站和入站是两个不同的方向。

    请注意,为了程序简单，我们的编码器和解码器不检查任何值的最大大小。在实际实现中你需要一些验证检查,如果检测到违反协议，则使用 EncoderException 或 DecoderException(或一个子类)。

    实现 Memcached 编码器

    本节我们将简要介绍编码器的实现。正如我们提到的,编码器负责编码消息为字节序列。这些字节可以通过网络发送到远端。为了发送请求，我们首先创建 MemcachedRequest 类,稍后编码器实现会编码为一系列字节。下面的清单显示了我们的 MemcachedRequest 类

    Listing 14.1 Implementation of a Memcached request

    1. public class MemcachedRequest { //1
    2.     private static final Random rand = new Random();
    3.     private final int magic = 0x80;//fixed so hard coded
    4.     private final byte opCode; //the operation e.g. set or get
    5.     private final String key; //the key to delete, get or set
    6.     private final int flags = 0xdeadbeef; //random
    7.     private final int expires; //0 = item never expires
    8.     private final String body; //if opCode is set, the value
    9.     private final int id = rand.nextInt(); //Opaque
    10.     private final long cas = 0; //data version check...not used
    11.     private final boolean hasExtras; //not all ops have extras
    13.     public MemcachedRequest(byte opcode, String key, String value) {
    14.         this.opCode = opcode;
    15.         this.key = key;
    16.         this.body = value == null ? "" : value;
    17.         this.expires = 0;
    18.         //only set command has extras in our example
    19.         hasExtras = opcode == Opcode.SET;
    20.     }
    22.     public MemcachedRequest(byte opCode, String key) {
    23.         this(opCode, key, null);
    24.     }
    26.     public int magic() { //2
    27.         return magic;
    28.     }
    30.     public int opCode() {  //3
    31.         return opCode;
    32.     }
    34.     public String key() {  //4
    35.         return key;
    36.     }
    38.     public int flags() {  //5
    39.         return flags;
    40.     }
    42.     public int expires() {  //6
    43.         return expires;
    44.     }
    46.     public String body() {  //7
    47.         return body;
    48.     }
    50.     public int id() {  //8
    51.         return id;
    52.     }
    54.     public long cas() {  //9
    55.         return cas;
    56.     }
    58.     public boolean hasExtras() {  //10
    59.         return hasExtras;
    60.     }
    61. }
    1. 这个类将会发送请求到 Memcached server
    2. 幻数，它可以用来标记文件或者协议的格式
    3. opCode,反应了响应的操作已经创建了
    4. 执行操作的 key
    5. 使用的额外的 flag
    6. 表明到期时间
    7. body
    8. 请求的 id。这个id将在响应中回显。
    9. compare-and-check 的值
    10. 如果有额外的使用，将返回 true

    你如果想实现 Memcached 的其余部分协议,你只需要将 client.op(op 任何新的操作添加)转换为其中一个方法请求。我们需要两个更多的支持类,在下一个清单所示

    Listing 14.2 Possible Memcached operation codes and response statuses

    1. public class Status {
    2.     public static final short NO_ERROR = 0x0000;
    3.     public static final short KEY_NOT_FOUND = 0x0001;
    4.     public static final short KEY_EXISTS = 0x0002;
    5.     public static final short VALUE_TOO_LARGE = 0x0003;
    6.     public static final short INVALID_ARGUMENTS = 0x0004;
    7.     public static final short ITEM_NOT_STORED = 0x0005;
    8.     public static final short INC_DEC_NON_NUM_VAL = 0x0006;
    9. }
    10. public class Opcode {
    11.     public static final byte GET = 0x00;
    12.     public static final byte SET = 0x01;
    13.     public static final byte DELETE = 0x04;
    14. }

    一个 Opcode 告诉 Memcached 要执行哪些操作。每个操作都由一个字节表示。同样的,当 Memcached 响应一个请求,响应头中包含两个字节代表响应状态。状态和 Opcode 类表示这些 Memcached 的构造。这些操作码可以使用当你构建一个新的 MemcachedRequest 指定哪个行动应该由它引发的。

    但现在可以集中精力在编码器上：

    Listing 14.3 MemcachedRequestEncoder implementation

    1. public class MemcachedRequestEncoder extends
    2.         MessageToByteEncoder<MemcachedRequest> { //1
    3.     @Override
    4.     protected void encode(ChannelHandlerContext ctx, MemcachedRequest msg,
    5.                           ByteBuf out) throws Exception {  //2
    6.         byte[] key = msg.key().getBytes(CharsetUtil.UTF_8);
    7.         byte[] body = msg.body().getBytes(CharsetUtil.UTF_8);
    8.         //total size of the body = key size + content size + extras size   //3
    9.         int bodySize = key.length + body.length + (msg.hasExtras() ? 8 : 0);
    11.         //write magic byte  //4
    12.         out.writeByte(msg.magic());
    13.         //write opcode byte  //5
    14.         out.writeByte(msg.opCode());
    15.         //write key length (2 byte) //6
    16.         out.writeShort(key.length); //key length is max 2 bytes i.e. a Java short  //7
    17.         //write extras length (1 byte)
    18.         int extraSize =  msg.hasExtras() ? 0x08 : 0x0;
    19.         out.writeByte(extraSize);
    20.         //byte is the data type, not currently implemented in Memcached but required //8
    21.         out.writeByte(0);
    22.         //next two bytes are reserved, not currently implemented but are required  //9
    23.         out.writeShort(0);
    25.         //write total body length ( 4 bytes - 32 bit int)  //10
    26.         out.writeInt(bodySize);
    27.         //write opaque ( 4 bytes)  -  a 32 bit int that is returned in the response //11
    28.         out.writeInt(msg.id());
    30.         //write CAS ( 8 bytes)
    31.         out.writeLong(msg.cas());   //24 byte header finishes with the CAS  //12
    33.         if (msg.hasExtras()) {
    34.             //write extras (flags and expiry, 4 bytes each) - 8 bytes total  //13
    35.             out.writeInt(msg.flags());
    36.             out.writeInt(msg.expires());
    37.         }
    38.         //write key   //14
    39.         out.writeBytes(key);
    40.         //write value  //15
    41.         out.writeBytes(body);
    42.     }
    43. }
    1. 该类是负责编码 MemachedRequest 为一系列字节
    2. 转换的 key 和实际请求的 body 到字节数组
    3. 计算 body 大小
    4. 写幻数到 ByteBuf 字节
    5. 写 opCode 作为字节
    6. 写 key 长度z作为 short
    7. 编写额外的长度作为字节
    8. 写数据类型,这总是0,因为目前不是在 Memcached,但可用于使用
      后来的版本
    9. 为保留字节写为 short ,后面的 Memcached 版本可能使用
    10. 写 body 的大小作为 long
    11. 写 opaque 作为 int
    12. 写 cas 作为 long。这个是头文件的最后部分，在 body 的开始
    13. 编写额外的 flag 和到期时间为 int
    14. 写 key
    15. 这个请求完成后 写 body。

    总结，编码器 使用 Netty 的 ByteBuf 处理请求，编码 MemcachedRequest 成一套正确排序的字节。详细步骤为：

    • 写幻数字节。
    • 写 opcode 字节。
    • 写 key 长度(2字节)。
    • 写额外的长度(1字节)。
    • 写数据类型(1字节)。
    • 为保留字节写 null 字节(2字节)。
    • 写 body 长度(4字节- 32位整数)。
    • 写 opaque(4个字节,一个32位整数在响应中返回)。
    • 写 CAS(8个字节)。
    • 写 额外的(flag 和 到期,4字节)= 8个字节
    • 写 key
    • 写 值

    无论你放入什么到输出缓冲区( 调用 ByteBuf) Netty 的将向服务器发送被写入请求。下一节将展示如何进行反向通过解码器工作。

    实现 Memcached 解码器

    将 MemcachedRequest 对象转为字节序列，Memcached 仅需将字节转到响应对象返回即可。

    先见一个 POJO:

    Listing 14.7 Implementation of a MemcachedResponse

    1. public class MemcachedResponse {  //1
    2.     private final byte magic;
    3.     private final byte opCode;
    4.     private byte dataType;
    5.     private final short status;
    6.     private final int id;
    7.     private final long cas;
    8.     private final int flags;
    9.     private final int expires;
    10.     private final String key;
    11.     private final String data;
    13.     public MemcachedResponse(byte magic, byte opCode,
    14.             byte dataType,                             short status, 
    15.             int id, long cas,
    16.             int flags, int expires, String key, String data) {
    17.         this.magic = magic;
    18.         this.opCode = opCode;
    19.         this.dataType = dataType;
    20.         this.status = status;
    21.         this.id = id;
    22.         this.cas = cas;
    23.         this.flags = flags;
    24.         this.expires = expires;
    25.         this.key = key;
    26.         this.data = data;
    27.     }
    29.     public byte magic() { //2
    30.         return magic;
    31.     }
    33.     public byte opCode() { //3
    34.         return opCode;
    35.     }
    37.     public byte dataType() { //4
    38.         return dataType;
    39.     }
    41.     public short status() {  //5
    42.         return status;
    43.     }
    45.     public int id() {  //6
    46.         return id;
    47.     }
    49.     public long cas() {  //7
    50.         return cas;
    51.     }
    53.     public int flags() {  //8
    54.         return flags;
    55.     }
    57.     public int expires() { //9
    58.         return expires;
    59.     }
    61.     public String key() {  //10
    62.         return key;
    63.     }
    65.     public String data() {  //11
    66.         return data; 
    67.     }
    68. }
    1. 该类,代表从 Memcached 服务器返回的响应
    2. 幻数
    3. opCode,这反映了创建操作的响应
    4. 数据类型,这表明这个是基于二进制还是文本
    5. 响应的状态,这表明如果请求是成功的
    6. 惟一的 id
    7. compare-and-set 值
    8. 使用额外的 flag
    9. 表示该值存储的一个有效期
    10. 响应创建的 key
    11. 实际数据

    下面为 MemcachedResponseDecoder， 使用了 ByteToMessageDecoder 基类，用于将 字节序列转为 MemcachedResponse

    Listing 14.4 MemcachedResponseDecoder class

    1. public class MemcachedResponseDecoder extends ByteToMessageDecoder {  //1
    2.     private enum State {  //2
    3.         Header,
    4.         Body
    5.     }
    7.     private State state = State.Header;
    8.     private int totalBodySize;
    9.     private byte magic;
    10.     private byte opCode;
    11.     private short keyLength;
    12.     private byte extraLength;
    13.     private short status;
    14.     private int id;
    15.     private long cas;
    17.     @Override
    18.     protected void decode(ChannelHandlerContext ctx, ByteBuf in,
    19.                           List<Object> out) { 
    20.         switch (state) { //3
    21.             case Header:
    22.                 if (in.readableBytes() < 24) {
    23.                     return;//response header is 24  bytes  //4
    24.                 }
    25.                 magic = in.readByte();  //5
    26.                 opCode = in.readByte();
    27.                 keyLength = in.readShort();
    28.                 extraLength = in.readByte();
    29.                 in.skipBytes(1);
    30.                 status = in.readShort();
    31.                 totalBodySize = in.readInt();
    32.                 id = in.readInt(); //referred to in the protocol spec as opaque
    33.                 cas = in.readLong();
    35.                 state = State.Body;
    36.             case Body:
    37.                 if (in.readableBytes() < totalBodySize) {
    38.                     return; //until we have the entire payload return  //6
    39.                 }
    40.                 int flags = 0, expires = 0;
    41.                 int actualBodySize = totalBodySize;
    42.                 if (extraLength > 0) {  //7
    43.                     flags = in.readInt();
    44.                     actualBodySize -= 4;
    45.                 }
    46.                 if (extraLength > 4) {  //8
    47.                     expires = in.readInt();
    48.                     actualBodySize -= 4;
    49.                 }
    50.                 String key = "";
    51.                 if (keyLength > 0) {  //9
    52.                     ByteBuf keyBytes = in.readBytes(keyLength);
    53.                     key = keyBytes.toString(CharsetUtil.UTF_8);
    54.                     actualBodySize -= keyLength;
    55.                 }
    56.                 ByteBuf body = in.readBytes(actualBodySize);  //10
    57.                 String data = body.toString(CharsetUtil.UTF_8);
    58.                 out.add(new MemcachedResponse(  //1
    59.                         magic,
    60.                         opCode,
    61.                         status,
    62.                         id,
    63.                         cas,
    64.                         flags,
    65.                         expires,
    66.                         key,
    67.                         data
    68.                 ));
    70.                 state = State.Header;
    71.         }
    73.     }
    74. }
    1. 类负责创建的 MemcachedResponse 读取字节
    2. 代表当前解析状态,这意味着我们需要解析的头或 body
    3. 根据解析状态切换
    4. 如果不是至少24个字节是可读的,它不可能读整个头部,所以返回这里,等待再通知一次数据准备阅读
    5. 阅读所有头的字段
    6. 检查是否足够的数据是可读用来读取完整的响应的 body。长度是从头读取
    7. 检查如果有任何额外的 flag 用于读，如果是这样做
    8. 检查如果响应包含一个 expire 字段，有就读它
    9. 检查响应是否包含一个 key ,有就读它
    10. 读实际的 body 的 payload
    11. 从前面读取字段和数据构造一个新的 MemachedResponse

    所以在实现发生了什么事?我们知道一个 Memcached 响应有24位头;我们不知道是否所有数据,响应将被包含在输入 ByteBuf ，当解码方法调用时。这是因为底层网络堆栈可能将数据分解成块。所以确保我们只解码当我们有足够的数据,这段代码检查是否可用可读的字节的数量至少是24。一旦我们有24个字节,我们可以确定整个消息有多大,因为这个信息包含在24位头。

    当我们解码整个消息,我们创建一个 MemcachedResponse 并将其添加到输出列表。任何对象添加到该列表将被转发到下一个ChannelInboundHandler 在 ChannelPipeline,因此允许处理。