写点什么

Netty 学习之旅 ------ 高仿 Dubbo 服务调用模型、私有协议实现、编码解码器使用实践

  • 2022 年 4 月 23 日
  • 本文字数:3871 字

    阅读完需:约 13 分钟

  • @param in //当前累积缓存区字节

  • @param out //解码后的消息


*/


@Override


protected void decode(ChannelHandlerContext ctx, ByteBuf in,


List<Object> out) throws Exception {


// TODO Auto-generated method stub


LogUtils.log("RequestFrameDecoder start calling....");


int r = in.readableBytes();


LogUtils.log("RequestFrameDecoder 累计缓存去大小:%d", r);


if (r < HEAD_LENGTH) {


LogUtils.log("RequestFrameDecoder 当前累积缓存区可读数据长度为 %d,小于协议头的长度 %d", r, HEAD_LENGTH);


return;


}


int currentReaderIndex = in.readerIndex();


int requestId = in.getInt(currentReaderIndex);// 0-3


short requestType = in.getShort(currentReaderIndex + 4);// 4-5


short contentLength = in.getShort(currentReaderIndex + 6);// 6-7 头部信息共 8 个字节


LogUtils.log("RequestFrameDecoder requestId:%d,requestType:%d,contentLength:%d", requestId, requestType, contentLength);


Request request = null;


if (requestType == 1) {


if (r < (HEAD_LENGTH + contentLength)) { // 如果累积缓存区可读容量小于一个请求包的长度,直接返回


LogUtils.log("RequestFrameDecoder 当前累积缓存区可读数据长度为 %d,小于数据报的长度 %d", r, HEAD_LENGTH + contentLength);


return;


}


// 否则,直接读取数据


in.skipBytes(HEAD_LENGTH);// 先跳过协议头信息


int slen = in.readShort();


byte[] serviceData = new byte[slen];


in.readBytes(serviceData, 0, slen);


short plen = in.readShort();


byte[] paramsData = new byte[plen];


in.readBytes(paramsData, 0, plen);


out.add(request = new Request(requestId, requestType,


new String(serviceData), new String(paramsData)));


request.setContentLength(contentLength); //这里,其实上层应用不需要知道请求包的长度,这里属于多于的操作


LogUtils.log("RequestFrameDecoder 成功从服务端响应中解析出业务包,requestId:%d", requestId);


} else if (requestType == 2) { // 此处不对请求类型进行校验


in.skipBytes(HEAD_LENGTH);


out.add(request = new Request(requestId, requestType));


request.setContentLength(contentLength); //这里,其实上层应用不需要知道请求包的长度,这里属于多于的操作


LogUtils.log("RequestFrameDecoder 成功从服务端响应中解析出心跳包,requestId:%d", requestId);


}


}


}


package persistent.prestige.demo.netty.dubbo.net;


import io.netty.buffer.ByteBuf;


import io.netty.channel.ChannelHandlerContext;


import io.netty.handler.codec.MessageToByteEncoder;


import persistent.prestige.demo.netty.dubbo.utils.LogUtils;


/**


  • 请求消息编码器,实现自定义协议,用于客户端

  • 自定义协议如下

  • 4 字节的请求序列号 2 字节的请求类型 2 字节数据长度 2 字节服务调用长度 N 字节的服务调用名 2 字节请求参数长度 请求参数

  • 其中请求类型为 1:服务调用 2:心跳包,如果是心跳包的话,只有协议头,也就是【4 字节请求序列号,2 字节请求类型,2 字节数据长度】

  • @author prestigeding@126.com


*/


public class RequestFrameEncoder extends MessageToByteEncoder<Request> {


@Override


protected void encode(ChannelHandlerContext ctx, Request msg, ByteBuf out)


throws Exception {


LogUtils.log("RequestFrameEncoder begin calling...requestId:%d,requestType:%d.",msg.getRequestId(), msg.getRequestType());


out.writeInt(msg.getRequestId()); // requestId


out.writeShort(msg.getRequestType()); // requestType


if(msg.getRequestType() == 1 ) { //服务调用


byte[] serviceData = msg.getService().getBytes();


byte[] paramsData = msg.getParams().getBytes();


int slen, plen;


out.writeShort((


2 + (slen = serviceData.length) +


2 + (plen = paramsData.length)));


out.writeShort(slen);


out.writeBytes(serviceData);


out.writeShort(plen);


out.writeBytes(paramsData);


} else if(msg.getRequestType() == 2 ) { //心跳包


out.writeShort(0);


}


LogUtils.log("RequestFrameEncoder end call....");


}


}


package persistent.prestige.demo.netty.dubbo.net;


import java.util.List;


import io.netty.buffer.ByteBuf;


import io.netty.channel.ChannelHandlerContext;


import io.netty.handler.codec.ByteToMessageDecoder;


import persistent.prestige.demo.netty.dubbo.utils.LogUtils;


/**


  • 响应信息解码器 用于客户端

  • @author dingwei2

  • 响应报文格式

  • 4 字节的请求序列号 2 字节的执行状态码 2 字节数据长度 多字节的响应数据信息(data)


*/


public class ResponseFrameDecoder extends ByteToMessageDecoder {


private static final int HEAD_LENGTH = 8;


@Override


protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {


LogUtils.log("ResponseFrameDecoder start calling....");


// TODO Auto-generated method stub


int r = in.readableBytes();


LogUtils.log("ResponseFrameDecoder 累积缓存区中可读长度:%d", r);


if(r < HEAD_LENGTH ) {


LogUtils.log("ResponseFrameDecoder 当前累积缓存区可读数据长度为 %d,小于协议头的长度 %d", r, HEAD_LENGTH);


return;


}


int currentReaderIndex = 《一线大厂 Java 面试题解析+后端开发学习笔记+最新架构讲解视频+实战项目源码讲义》无偿开源 威信搜索公众号【编程进阶路】 in.readerIndex();


int requestId = in.getInt(currentReaderIndex); // 0-3 4 个字节


short code = in.getShort(currentReaderIndex + 4); // 4,5


short contentLength = in.getShort(currentReaderIndex + 6); //响应数据长度


LogUtils.log("ResponseFrameDecoder requestId:%d,code:%d,contentLength:%d", requestId, code, contentLength);


int packetLen;


if( r < (packetLen = HEAD_LENGTH + contentLength)) {


LogUtils.log("ResponseFrameDecoder 可读长度:%d,需要包长度:%d", r, packetLen);


return;


}


in.skipBytes( HEAD_LENGTH );


String data;


if(contentLength > 0 ) {


byte[] dst = new byte[contentLength];


in.readBytes(dst, 0, contentLength);


data = new String(dst);


} else {


data = "";


}


out.add(new Response(requestId, code, new String(data)));


LogUtils.log("ResponseFrameDecoder 成功解析服务端响应信息并准备给下游处理器处理");


}


}


package persistent.prestige.demo.netty.dubbo.net;


import io.netty.buffer.ByteBuf;


import io.netty.channel.ChannelHandlerContext;


import io.netty.handler.codec.MessageToByteEncoder;


import persistent.prestige.demo.netty.dubbo.utils.LogUtils;


/**


  • 响应信息 编码器,,用于服务器

  • @author dingwei2

  • 响应报文格式

  • 4 字节的请求序列号 2 字节的执行状态码 2 字节数据长度 多字节的响应数据信息(data)


*/


public class ResponseFrameEncoder extends MessageToByteEncoder<Response> {


@Override


protected void encode(ChannelHandlerContext ctx, Response msg, ByteBuf out) throws Exception {


// TODO Auto-generated method stub


LogUtils.log("ResponseFrameEncoder:服务端编码响应,requestId:%d,code:%d,data:%s", msg.getRequestId(), msg.getCode(), msg.getData());


out.writeInt(msg.getRequestId());


out.writeShort(msg.getCode());


byte[] data = msg.getData().getBytes();


out.writeShort(data.length);


out.writeBytes(data);


LogUtils.log("ResponseFrameEncoder:服务端编码响应成功");


}


}


package persistent.prestige.demo.netty.dubbo.net;


import java.util.concurrent.ConcurrentHashMap;


import java.util.concurrent.TimeUnit;


import java.util.concurrent.atomic.AtomicInteger;


import io.netty.bootstrap.Bootstrap;


import io.netty.channel.ChannelFuture;


import io.netty.channel.ChannelHandlerAdapter;


import io.netty.channel.ChannelHandlerContext;


import io.netty.channel.ChannelInitializer;


import io.netty.channel.nio.NioEventLoopGroup;


import io.netty.channel.socket.SocketChannel;


import io.netty.channel.socket.nio.NioSocketChannel;


import persistent.prestige.demo.netty.dubbo.net.heart.HeartExecutorThreadPool;


import persistent.prestige.demo.netty.dubbo.net.heart.HeartTask;


import persistent.prestige.demo.netty.dubbo.utils.LogUtils;


/**


  • netty 客户端 connection 一个 NettyConnection 就是一条长连接

  • @author prestigeding@126.com


*/


public class NettyClientConnection {


private String serverIp;


private int serverPort;


private boolean connect = false;


private Bootstrap bootstrap;


private NioEventLoopGroup workGroup;


private ChannelFuture channelFuture;


//客户端连接请求


private static final ConcurrentHashMap<Integer, ResponseFuture> requestMap = new ConcurrentHashMap<Integer, ResponseFuture>();


private static final AtomicInteger seq = new AtomicInteger(0);


private HeartExecutorThreadPool heartExecutor = null;


/**


  • @param serverIp

  • @param serverPort


*/


public NettyClientConnection(String serverIp, int serverPort) {


this.serverIp = serverIp;


this.serverPort = serverPort;


workGroup = new NioEventLoopGroup();


bootstrap = new Bootstrap();


bootstrap.group(workGroup)


.channel(NioSocketChannel.class)

用户头像

还未添加个人签名 2022.04.13 加入

还未添加个人简介

评论

发布
暂无评论
Netty学习之旅------高仿Dubbo服务调用模型、私有协议实现、编码解码器使用实践_Java_爱好编程进阶_InfoQ写作社区