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