2021-05-05 16:51  阅读(139)
文章分类:Netty 学习之旅 文章标签:NettyNetty 学习
©  原文作者:中间件兴趣圈 原文地址:https://blog.csdn.net/prestigeding/article/details/53977445

1、本文实现如下功能

1、客户端与服务端基于单一长连接进行通信,客户端一条连接被多个线程使用。

2、实现私有协议

请求协议 :

协议头:4字节的请求序列号 2字节的请求类型 2字节数据包长度

数据部分:

服务调用:2字节请求服务名长度 若干字节请求服务名 2字节请求参数长度 若干字节参数。

心跳包:数据部分无。

响应信息

协议头:4字节的请求序列号 2字节的执行状态码 2字节数据长度。

数据部分: 多字节的响应数据信息(data)。

3、代码基于Netty5

几个核心类代码如下:

        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;
    
        /**
         * 请求消息   解码器,实现自定义协议,,,用于服务端
         * 
         * 自定义协议如下 4字节的请求序列号 2字节的请求类型 2字节数据长度 2字节服务调用长度 N字节的服务调用名 2字节请求参数长度 请求参数
         * 其中请求类型为1:服务调用2:心跳包,如果是心跳包的话,只有协议头,也就是【4字节请求序列号,2字节请求类型,2字节数据长度】
         * 
         * @author prestigeding@126.com
         *
         */
        public class RequestFrameDecoder extends ByteToMessageDecoder {
    
            /** 协议头 长度 */
            private static final int HEAD_LENGTH = 8;
    
            /*
             * (non-Javadoc)
             * 
             * @param ctx Handler执行环境
             * 
             * @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 = 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)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch)
                                    throws Exception {
                                // TODO Auto-generated method stub
                                ch.pipeline().addLast(new ResponseFrameDecoder())
                                             .addLast(new RequestFrameEncoder())
                                             .addLast(new DispatchHandler());
    
                            }
    
                        });
    
            }
    
            private void connect() {
                if(!connect) {
    
                    synchronized (this) {
                        try {
                            if(!connect) {
                                channelFuture = bootstrap.connect(serverIp, serverPort).sync();
                                heartExecutor = new HeartExecutorThreadPool(0, this, 3);
                                connect = true;
    
                                heartExecutor.scheduleAtFixedRate(new HeartTask(this), 1, 3, TimeUnit.SECONDS);//单位为秒
                            }
    
                        } catch (InterruptedException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                            throw new NullPointerException("连接服务端失败");
                        }
                    }
    
                }
            }
    
            /**
             * 发送服务调用
             * @param service
             * @param params
             * @return
             */
            public ResponseFuture sendMsg(String service, String params) {
                Integer requestId = seq.incrementAndGet();
                System.out.println("客户端requestId:" + requestId);
                return send(new Request(requestId, (short)1, service, params));
            }
    
            /**
             * 发送心跳包
             * @return
             */
            public ResponseFuture sendHeatMsg() {
                Integer requestId = seq.incrementAndGet();
                return send(new Request(requestId, (short)2));
            }
    
            private ResponseFuture send(Request request) {
                connect();
                ResponseFuture responseFuture = new ResponseFuture();
                requestMap.putIfAbsent(request.getRequestId(), responseFuture);
                channelFuture.channel().writeAndFlush(request);
                return responseFuture;
            }
    
            /**
             * 关闭连接
             */
            public void close() {
                try {
                    channelFuture.channel().close().sync();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } finally {
                    workGroup.shutdownGracefully();
                    if(heartExecutor != null) {
                        heartExecutor.shutdownNow();
                    }
                }
            }
    
            /**
             * 客户端命令转发器
             * @author prestigeding@126.com
             *
             */
            private class DispatchHandler extends ChannelHandlerAdapter {
    
                /**
                 * 得到服务端的响应后,放入结果中,供客户端使用结果
                 */
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg)
                        throws Exception {
                    LogUtils.log("客户端收到服务端响应,并开始处理数据");
                    Response response = (Response)msg;
                    LogUtils.log("服务端响应requestId:", response.getRequestId());
                    ResponseFuture f = requestMap.remove(response.getRequestId());
                    if(f != null) {
                        f.setResult(response);
                    } else {
                        LogUtils.log("警告:客户端丢失请求信息");
                    }
    
                }
    
            }
    
        }

完整代码如下:http://download.csdn.net/detail/prestigeding/9775190

其中Server是服务器启动类,Client 是客户端启动类。


来源:https://blog.csdn.net/prestigeding/article/details/53977445

点赞(0)
版权归原创作者所有,任何形式转载请联系作者; Java 技术驿站 >> 高仿Dubbo服务调用模型、私有协议实现、编码解码器使用实践
上一篇
写事件处理NioSocketChannel、ChannelOutbondBuffer源码分析