博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
简易RPC框架:基于 netty 的协议编解码
阅读量:6083 次
发布时间:2019-06-20

本文共 9263 字,大约阅读时间需要 30 分钟。

概述

在这篇文章中已经给出了协议的具体细节,协议类型为二进制协议,如下:

------------------------------------------------------------------------ | 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 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
{ @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); }}

转载地址:http://smkwa.baihongyu.com/

你可能感兴趣的文章
李开复给大支招 大学生创业有五忌
查看>>
大学学习有感
查看>>
CSS固定DIV,导航条顶部固定fixed(兼容IE6)
查看>>
docker 容器创建参数错误记录
查看>>
python3 笔记
查看>>
kali linux下的网络配置
查看>>
我的友情链接
查看>>
Windows系统下的TCP参数优化(注册表\TCPIP\Parameters)
查看>>
资源共享开源站
查看>>
Open×××中TAP-win32d的net30问题
查看>>
Linux常用命令总结之(九)tail
查看>>
【Glassfish调查】获取客户端Addr和Host
查看>>
2011年山东济南java开发定向就业班招生
查看>>
磁盘格式化与管理知识点总结
查看>>
如何在ASA防火墙上实现ipsec ***
查看>>
前端基础---jquery练习
查看>>
保持健康,远离疾病
查看>>
如何推断乱码是从何种编码转换到何种编码导致的乱码?
查看>>
Objective-C语法之动态类型(isKindOfClass, isMemberOfClass,id)等
查看>>
【SMTP 补录】
查看>>