简易RPC框架:基于 netty 的协议编解码

Java框架

浏览数:175

2018-12-8

概述

《简易RPC框架:需求与设计》这篇文章中已经给出了协议的具体细节,协议类型为二进制协议,如下:

 ------------------------------------------------------------------------
 | magic (2bytes) | version (1byte) |  type (1byte)  | reserved (7bits) | 
 ------------------------------------------------------------------------
 | status (1byte) |    id (8bytes)    |        body length (4bytes)     |
 ------------------------------------------------------------------------
 |                                                                      |
 |                   body ($body_length bytes)                          |
 |                                                                      |
 ------------------------------------------------------------------------

协议的解码我们称为 decode,编码我们成为 encode,下文我们将直接使用 decode 和 encode 术语。

decode 的本质就是讲接收到的一串二进制报文,转化为具体的消息对象,在 Java 中,就是将这串二进制报文所包含的信息,用某种类型的对象存储起来。

encode 则是将存储了信息的对象,转化为具有相同含义的一串二进制报文,然后网络收发模块再将报文发出去。

无论是 rpc 客户端还是服务端,都需要有一个 decode 和 encode 的逻辑。

消息类型

rpc 客户端与服务端之间的通信,需要通过发送不同类型的消息来实现,例如:client 向 server 端发送的消息,可能是请求消息,可能是心跳消息,可能是认证消息,而 server 向 client 发送的消息,一般就是响应消息。

利用 Java 中的枚举类型,可以将消息类型进行如下定义:

/**
 * 消息类型
 *
 * @author beanlam
 * @version 1.0  
 */ 

public enum MessageType {

    REQUEST((byte) 0x01), HEARTBEAT((byte) 0x02), CHECKIN((byte) 0x03), RESPONSE(
            (byte) 0x04), UNKNOWN((byte) 0xFF);

    private byte code;

    MessageType(byte code) {
        this.code = code;
    }

    public static MessageType valueOf(byte code) {
        for (MessageType instance : values()) {
            if (instance.code == code) {
                return instance;
            }
        }
        return UNKNOWN;
    }

    public byte getCode() {
        return code;
    }
}

在这个类中设计了 valueOf 方法,方便进行具体的 byte 字节与具体的消息枚举类型之间的映射和转换。

调用状态设计

client 主动发起的一次 rpc 调用,要么成功,要么失败,server 端有责任告知 client 此次调用的结果,client 也有责任去感知调用失败的原因,因为不一定是 server 端造成的失败,可能是因为 client 端在对消息进行预处理的时候,例如序列化,就已经出错了,这种错误也应该作为一次调用的调用结果返回给 client 调用者。因此引入一个调用状态,与消息类型一样,它也借助了 Java 语言里的枚举类型来实现,并实现了方便的 valueOf 方法:

/**

 * 调用状态

 *

 * @author beanlam

 * @version 1.0

 */

public enum InvocationStatus {

    OK((byte) 0x01), CLIENT_TIMEOUT((byte) 0x02), SERVER_TIMEOUT(

            (byte) 0x03), BAD_REQUEST((byte) 0x04), BAD_RESPONSE(

            (byte) 0x05), SERVICE_NOT_FOUND((byte) 0x06), SERVER_SERIALIZATION_ERROR(

            (byte) 0x07), CLIENT_SERIALIZATION_ERROR((byte) 0x08), CLIENT_CANCELED(

            (byte) 0x09), SERVER_BUSY((byte) 0x0A), CLIENT_BUSY(

            (byte) 0x0B), SERIALIZATION_ERROR((byte) 0x0C), INTERNAL_ERROR(

            (byte) 0x0D), SERVER_METHOD_INVOKE_ERROR((byte) 0x0E), UNKNOWN((byte) 0xFF);

    private byte code;

    InvocationStatus(byte code) {

        this.code = code;

    }

    public static InvocationStatus valueOf(byte code) {

        for (InvocationStatus instance : values()) {

            if (code == instance.code) {

                return instance;

            }

        }

        return UNKNOWN;

    }

    public byte getCode() {

        return code;

    }

}

消息实体设计

我们将 client 往 server 端发送的统一称为 rpc 请求消息,一个请求对应着一个响应,因此在 client 和 server 端间流动的信息大体上其实就只有两种,即要么是请求,要么是响应。我们将会定义两个类,分别是 RpcRequest 和 RpcResponse 来代表请求消息和响应消息。

另外由于无论是请求消息还是响应消息,它们都有一些共同的属性,例如说“调用上下文ID”,或者消息类型。因此会再定义一个 RpcMessage 类,作为父类。

消息类的关系

RpcMessage

/**

 * rpc消息

 *

 * @author beanlam

 * @version 1.0

 */

public class RpcMessage {

    private MessageType type;

    private long contextId;

    private Object data;

    public long getContextId() {

        return this.contextId;

    }

    public void setContextId(long id) {

        this.contextId = id;

    }

    public Object getData() {

        return this.data;

    }

    public void setData(Object data) {

        this.data = data;

    }

    public void setType(byte code) {

        this.type = MessageType.valueOf(code);

    }

    public MessageType getType() {

        return this.type;

    }

    public void setType(MessageType type) {

        this.type = type;

    }

    @Override

    public String toString() {

        return "[messageType=" + type.name() + ", contextId=" + contextId + ", data="

                + data + "]";

    }

}

RpcRequest

import java.util.concurrent.atomic.AtomicLong;

/**

 * rpc请求消息

 *

 * @author beanlam

 * @version 1.0

 */

public class RpcRequest extends RpcMessage {

    private static final AtomicLong ID_GENERATOR = new AtomicLong(0);

    public RpcRequest() {

        this(ID_GENERATOR.incrementAndGet());

    }

    public RpcRequest(long contextId) {

        setContextId(contextId);

        setType(MessageType.REQUEST);

    }

}

RpcResponse

/**

 *

 * rpc响应消息

 *

 * @author beanlam

 * @version 1.0

 */

public class RpcResponse extends RpcMessage {

    private InvocationStatus status = InvocationStatus.OK;

    public RpcResponse(long contextId) {

        setContextId(contextId);

        setType(MessageType.RESPONSE);

    }

    public InvocationStatus getStatus() {

        return this.status;

    }

    public void setStatus(InvocationStatus status) {

        this.status = status;

    }

    @Override

    public String toString() {

        return "RpcResponse[contextId=" + getContextId() + ", status=" + status.name() + "]";

    }

}

netty 编解码介绍

netty 是一个 NIO 框架,应该这么说,netty 是一个有良好设计思想的 NIO 框架。一个 NIO 框架必备的要素就是 reactor 线程模型,目前有一些比较优秀而且开源的小型 NIO 框架,例如分库分表中间件 mycat 实现的一个简易 NIO 框架,可以在这里看到。

netty 的主要特点有:微内核设计、责任链模式的业务逻辑处理、内存和资源泄露的检测等。其中编解码在 netty 中,都被设计成责任链上的一个一个 Handler。

decode 对于 netty 来说,它提供了 ByteToMessageDecoder,它也提供了 MessageToByteEncoder。

借助 netty 来实现协议编解码,实际上就是去在这两个handler里面实现编解码的逻辑。

decode

在实现 decode 逻辑时需要注意的一个问题是,由于二进制报文是在网络上发送的,因此一个完整的报文可能经过多个分组来发送的,什么意思呢,就是当有报文进来后,要确认报文是否完整,decode逻辑代码不能假设收到的报文就是一个完整报文,一般称这为“TCP半包问题”。同样,报文是连着报文发送的,意味着decode代码逻辑还要负责在一长串二进制序列中,分割出一个一个独立的报文,这称之为“TCP粘包问题”。

netty 本身有提供一些方便的 decoder handler 来处理 TCP 半包和粘包的问题。不过一般情况下我们不会直接去用它,因为我们的协议比较简单,自己在代码里处理一下就可以了。

完整的 decode 代码逻辑如下所示:

import cn.com.agree.ats.rpc.message.*;

import cn.com.agree.ats.util.logfacade.AbstractPuppetLoggerFactory;

import cn.com.agree.ats.util.logfacade.IPuppetLogger;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

/**

 * 协议解码器

 *

 * @author beanlam

 * @version 1.0

 */

public class ProtocolDecoder extends ByteToMessageDecoder {

    private static final IPuppetLogger logger = AbstractPuppetLoggerFactory

            .getInstance(ProtocolDecoder.class);

    private boolean magicChecked = false;

    @Override

    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> list)

            throws Exception {

        if (!magicChecked) {

            if (in.readableBytes() < ProtocolMetaData.MAGIC_LENGTH_IN_BYTES) {

                return;

            }

            magicChecked = true;

            if (!(in.getShort(in.readerIndex()) == ProtocolMetaData.MAGIC)) {

                logger.warn(

                        "illegal data received without correct magic number, channel will be close");

                ctx.close();

                magicChecked = false; //this line of code makes no any sense, but it's good for a warning

                return;

            }

        }

        if (in.readableBytes() < ProtocolMetaData.HEADER_LENGTH_IN_BYTES) {

            return;

        }

        int bodyLength = in

                .getInt(in.readerIndex() + ProtocolMetaData.BODY_LENGTH_OFFSET);

        if (in.readableBytes() < bodyLength + ProtocolMetaData.HEADER_LENGTH_IN_BYTES) {

            return;

        }

        magicChecked = false;// so far the whole packet was received

        in.readShort(); // skip the magic

        in.readByte(); // dont care about the protocol version so far

        byte type = in.readByte();

        byte status = in.readByte();

        long contextId = in.readLong();

        byte[] body = new byte[in.readInt()];

        in.readBytes(body);

        RpcMessage message = null;

        MessageType messageType = MessageType.valueOf(type);

        if (messageType == MessageType.RESPONSE) {

            message = new RpcResponse(contextId);

            ((RpcResponse) message).setStatus(InvocationStatus.valueOf(status));

        } else {

            message = new RpcRequest(contextId);

        }

        message.setType(messageType);

        message.setData(body);

        list.add(message);

    }

}

可以看到,我们解决半包问题的时候,是判断有没有收到我们期望收到的报文,如果没有,直接在 decode 方法里面 return,等有更多的报文被收到的时候,netty 会自动帮我们调起 decode 方法。而我们解决粘包问题的思路也很清晰,那就是一次只处理一个报文,不去动后面的报文内容。

还需要注意的是,在 netty 中,对于 ByteBuf 的 get 是不会消费掉报文的,而 read 是会消费掉报文的。当不确定报文是否收完整的时候,我们都是用 get开头的方法去试探性地验证报文是否接收完全,当确定报文接收完全后,我们才用 read 开头的方法去消费这段报文。

encode

直接贴代码,参考前文提到的协议格式阅读以下代码:

/**

 *

 * 协议编码器

 *

 * @author beanlam

 * @version 1.0

 */

public class ProtocolEncoder extends MessageToByteEncoder<RpcMessage> {

    @Override

    protected void encode(ChannelHandlerContext ctx, RpcMessage rpcMessage, ByteBuf out)

            throws Exception {

        byte status;

        byte[] data = (byte[]) rpcMessage.getData();

        if (rpcMessage instanceof RpcRequest) {

            RpcRequest request = (RpcRequest) rpcMessage;

            status = InvocationStatus.OK.getCode();

        } else {

            RpcResponse response = (RpcResponse) rpcMessage;

            status = response.getStatus().getCode();

        }

        out.writeShort(ProtocolMetaData.MAGIC);

        out.writeByte(ProtocolMetaData.VERSION);

        out.writeByte(rpcMessage.getType().getCode());

        out.writeByte(status);

        out.writeLong(rpcMessage.getContextId());

        out.writeInt(data.length);

        out.writeBytes(data);

    }

}

原文地址:https://segmentfault.com/a/1190000016924445