netty私有协议的开发

素颜马尾好姑娘i 2022-06-12 07:50 238阅读 0赞

通信协议从广义上区分，可以分为公有协议和私有协议。由于私有协议的灵活性，它往往会在某个公司或者组织内部使用，按需定制，也因为如此，升级起来会非常方便，灵活性好。绝大多数的私有协议传输层都基于TCP/IP，所以利用Netty的NIO TCP协议栈可以非常方便地进行私有协议的定制和开发。

## 私有协议介绍 ##

私有协议本质上是厂商内部发展和采用的标准，除非授权，其他厂商一般无权使用该协议。私有协议也称非标准协议，就是未经国际或国家标准化组织采纳或批准，由某个企业自己制订，协议实现细节不愿公开，只在企业自己生产的设备之间使用的协议。私有协议具有封闭性、垄断性、排他性等特点。如果网上大量存在私有（非标准）协议，现行网络或用户一旦使用了它，后进入的厂家设备就必须跟着使用这种非标准协议，才能够互连互通，否则根本不可能进入现行网络。这样，使用非标准协议的厂家就实现了垄断市场的愿望。

在传统的Java应用中，通常使用以下4种方式进行跨节点通信。

（1）通过RMI进行远程服务调用；

（2）通过Java的Socket+Java序列化的方式进行跨节点调用；

（3）利用一些开源的RPC框架进行远程服务调用，例如Facebook的Thrift，Apache的Avro等；

（4）利用标准的公有协议进行跨节点服务调用，例如HTTP+XML、RESTful+JSON或者WebService。

跨节点的远程服务调用，除了链路层的物理连接外，还需要对请求和响应消息进行编解码。在请求和应答消息本身以外，也需要携带一些其他控制和管理类指令，例如链路建立的握手请求和响应消息、链路检测的心跳消息等。当这些功能组合到一起之后，就会形成私有协议。

事实上，私有协议并没有标准的定义，只要是能够用于跨进程、跨主机数据交换的非标准协议，都可以称为私有协议。通常情况下，正规的私有协议都有具体的协议规范文档，类似于《XXXX协议VXX规范》，但是在实际的项目中，内部使用的私有协议往往是口头约定的规范，由于并不需要对外呈现或者被外部调用，所以一般不会单独写相关的内部私有协议规范文档。

## Netty协议栈功能设计 ##

使用Netty提供的异步TCP协议栈开发一个私有协议栈，该协议栈被命名为Netty协议栈。Netty协议栈用于内部各模块之间的通信，它基于TCP/IP协议栈，是一个类HTTP协议的应用层协议栈，相比于传统的标准协议栈，它更加轻巧、灵活和实用。

### 网络拓扑图 ###

在分布式组网环境下，每个Netty节点（Netty进程）之间建立长连接，使用Netty协议进行通信。Netty节点并没有服务端和客户端的区分，谁首先发起连接，谁就作为客户端，另一方自然就成为服务端。一个Netty节点既可以作为客户端连接另外的Netty节点，也可以作为Netty服务端被其他Netty节点连接，这完全取决于使用者的业务场景和具体的业务组网。

![990532-20161215161242683-1070510244.png][]

### 协议栈功能描述 ###

Netty协议栈承载了业务内部各模块之间的消息交互和服务调用，它的主要功能如下。

（1）基于Netty的NIO通信框架，提供高性能的异步通信能力；

（2）提供消息的编解码框架，可以实现POJO的序列化和反序列化；

（3）提供基于IP地址的白名单接入认证机制；

（4）链路的有效性校验机制；

（5）链路的断连重连机制。

### 通信模型  ###

![990532-20161215161437995-1502340572.png][]

（1）Netty协议栈客户端发送握手请求消息，携带节点ID等有效身份认证信息；

（2）Netty协议栈服务端对握手请求消息进行合法性校验，包括节点ID有效性校验、节点重复登录校验和IP地址合法性校验，校验通过后，返回登录成功的握手应答消息；

（3）链路建立成功之后，客户端发送业务消息；

（4）链路成功之后，服务端发送心跳消息；

（5）链路建立成功之后，客户端发送心跳消息；

（6）链路建立成功之后，服务端发送业务消息；

（7）服务端退出时，服务端关闭连接，客户端感知对方关闭连接后，被动关闭客户端连接。

备注：需要指出的是，Netty协议通信双方链路建立成功之后，双方可以进行全双工通信，无论客户端还是服务端，都可以主动发送请求消息给对方，通信方式可以是TWO WAY或者ONE WAY。双方之间的心跳采用Ping-Pong机制，当链路处于空闲状态时，客户端主动发送Ping消息给服务端，服务端接收到Ping消息后发送应答消息Pong给客户端，如果客户端连续发送N条Ping消息都没有接收到服务端返回的Pong消息，说明链路已经挂死或者对方处于异常状态，客户端主动关闭连接，间隔周期T后发起重连操作，直到重连成功。

### 消息定义 ###

Netty协议栈消息定义包含两部分：

1.  消息头；
2.  消息体。

![990532-20161215161904339-1672827031.png][]

![990532-20161215161915479-351556821.png][]

![990532-20161215161937261-1103661657.png][]

### Netty协议支持的字段类型 ###

![990532-20161215162023745-539949159.png][]

### Netty协议的编解码规范 ###

1.Netty协议的编码

Netty协议NettyMessage的编码规范如下:

（1）crcCode：java.nio.ByteBuffer.putInt(int value)，如果采用其他缓冲区实现，必须与其等价；

（2）length：java.nio.ByteBuffer.putInt(int value)，如果采用其他缓冲区实现，必须与其等价；

（3）sessionID：java.nio.ByteBuffer.putLong(long value)，如果采用其他缓冲区实现，必须与其等价；

（4）type: java.nio.ByteBuffer.put(byte b)，如果采用其他缓冲区实现，必须与其等价；

（5）priority：java.nio.ByteBuffer.put(byte b)，如果采用其他缓冲区实现，必须与其等价；

（6）attachment：它的编码规则为——如果attachment长度为0，表示没有可选附件，则将长度编码设为0，java.nio.ByteBuffer.putInt(0)；如果大于0，说明有附件需要编码，具体的编码规则如下:首先对附件的个数进行编码，java.nio.ByteBuffer.putInt(attachment.size())；然后对Key进行编码，再将它转换成byte数组之后编码内容.

（7）body的编码：通过JBoss Marshalling将其序列化为byte数组，然后调用java.nio.ByteBuffer.put(byte \[\] src)将其写入ByteBuffer缓冲区中。

由于整个消息的长度必须等全部字段都编码完成之后才能确认，所以最后需要更新消息头中的length字段，将其重新写入ByteBuffer中。

2.Netty协议的解码

相对于NettyMessage的编码，仍旧以java.nio.ByteBuffer为例，给出Netty协议的解码规范。

（1）crcCode：通过java.nio.ByteBuffer.getInt()获取校验码字段，其他缓冲区需要与其等价；

（2）length：通过java.nio.ByteBuffer.getInt()获取Netty消息的长度，其他缓冲区需要与其等价；

（3）sessionID：通过java.nio.ByteBuffer.getLong()获取会话ID，其他缓冲区需要与其等价；

（4）type：通过java.nio.ByteBuffer.get()获取消息类型，其他缓冲区需要与其等价；

（5）priority：通过java.nio.ByteBuffer.get()获取消息优先级，其他缓冲区需要与其等价；

（6）attachment：它的解码规则为——首先创建一个新的attachment对象，调用java.nio.ByteBuffer.getInt()获取附件的长度，如果为0，说明附件为空，解码结束，继续解消息体；如果非空，则根据长度通过for循环进行解码。

（7）body：通过JBoss的marshaller对其进行解码。

### 链路的建立 ###

Netty协议栈支持服务端和客户端，对于使用Netty协议栈的应用程序而言，不需要刻意区分到底是客户端还是服务端，在分布式组网环境中，一个节点可能既是服务端也是客户端，这个依据具体的用户场景而定。

Netty协议栈对客户端的说明如下：如果A节点需要调用B节点的服务，但是A和B之间还没有建立物理链路，则由调用方主动发起连接，此时，调用方为客户端，被调用方为服务端。

考虑到安全，链路建立需要通过基于IP地址或者号段的黑白名单安全认证机制，作为样例，本协议使用基于IP地址的安全认证，如果有多个IP，通过逗号进行分割。在实际商用项目中，安全认证机制会更加严格，例如通过密钥对用户名和密码进行安全认证。

客户端与服务端链路建立成功之后，由客户端发送握手请求消息，握手请求消息的定义如下。

（1）消息头的type字段值为3；

（2）可选附件为个数为0；

（3）消息体为空；

（4）握手消息的长度为22个字节。

服务端接收到客户端的握手请求消息之后，如果IP校验通过，返回握手成功应答消息给客户端，应用层链路建立成功。握手应答消息定义如下。

（1）消息头的type字段值为4；

（2）可选附件个数为0；

（3）消息体为byte类型的结果，0：认证成功；-1：认证失败。

链路建立成功之后，客户端和服务端就可以互相发送业务消息了。

### 链路的关闭 ###

由于采用长连接通信，在正常的业务运行期间，双方通过心跳和业务消息维持链路，任何一方都不需要主动关闭连接。

但是，在以下情况下，客户端和服务端需要关闭连接。

（1）当对方宕机或者重启时，会主动关闭链路，另一方读取到操作系统的通知信号，得知对方REST链路，需要关闭连接，释放自身的句柄等资源。由于采用TCP全双工通信，通信双方都需要关闭连接，释放资源；

（2）消息读写过程中，发生了I/O异常，需要主动关闭连接；

（3）心跳消息读写过程中发生了I/O异常，需要主动关闭连接；

（4）心跳超时，需要主动关闭连接；

（5）发生编码异常等不可恢复错误时，需要主动关闭连接。

### 可靠性设计 ###

Netty协议栈可能会运行在非常恶劣的网络环境中，网络超时、闪断、对方进程僵死或者处理缓慢等情况都有可能发生。为了保证在这些极端异常场景下Netty协议栈仍能够正常工作或者自动恢复，需要对它的可靠性进行统一规划和设计。

1.心跳机制

在凌晨等业务低谷期时段，如果发生网络闪断、连接被Hang住等网络问题时，由于没有业务消息，应用进程很难发现。到了白天业务高峰期时，会发生大量的网络通信失败，严重的会导致一段时间进程内无法处理业务消息。为了解决这个问题，在网络空闲时采用心跳机制来检测链路的互通性，一旦发现网络故障，立即关闭链路，主动重连。

具体的设计思路如下。

（1）当网络处于空闲状态持续时间达到T（连续周期T没有读写消息）时，客户端主动发送Ping心跳消息给服务端；

（2）如果在下一个周期T到来时客户端没有收到对方发送的Pong心跳应答消息或者读取到服务端发送的其他业务消息，则心跳失败计数器加1；

（3）每当客户端接收到服务的业务消息或者Pong应答消息，将心跳失败计数器清零；当连续N次没有接收到服务端的Pong消息或者业务消息，则关闭链路，间隔INTERVAL时间后发起重连操作；

（4）服务端网络空闲状态持续时间达到T后，服务端将心跳失败计数器加1；只要接收到客户端发送的Ping消息或者其他业务消息，计数器清零；

（5）服务端连续N次没有接收到客户端的Ping消息或者其他业务消息，则关闭链路，释放资源，等待客户端重连。

通过Ping-Pong双向心跳机制，可以保证无论通信哪一方出现网络故障，都能被及时地检测出来。为了防止由于对方短时间内繁忙没有及时返回应答造成的误判，只有连续N次心跳检测都失败才认定链路已经损害，需要关闭链路并重建链路。

当读或者写心跳消息发生I/O异常的时候，说明链路已经中断，此时需要立即关闭链路，如果是客户端，需要重新发起连接。如果是服务端，需要清空缓存的半包信息，等待客户端重连。

2.重连机制

如果链路中断，等待INTERVAL时间后，由客户端发起重连操作，如果重连失败，间隔周期INTERVAL后再次发起重连，直到重连成功。

为了保证服务端能够有充足的时间释放句柄资源，在首次断连时客户端需要等待INTERVAL时间之后再发起重连，而不是失败后就立即重连。

为了保证句柄资源能够及时释放，无论什么场景下的重连失败，客户端都必须保证自身的资源被及时释放，包括但不限于SocketChannel、Socket等。

重连失败后，需要打印异常堆栈信息，方便后续的问题定位。

3.重复登录保护

当客户端握手成功之后，在链路处于正常状态下，不允许客户端重复登录，以防止客户端在异常状态下反复重连导致句柄资源被耗尽。

服务端接收到客户端的握手请求消息之后，首先对IP地址进行合法性检验，如果校验成功，在缓存的地址表中查看客户端是否已经登录，如果已经登录，则拒绝重复登录，返回错误码-1，同时关闭TCP链路，并在服务端的日志中打印握手失败的原因。

客户端接收到握手失败的应答消息之后，关闭客户端的TCP连接，等待INTERVAL时间之后，再次发起TCP连接，直到认证成功。

为了防止由服务端和客户端对链路状态理解不一致导致的客户端无法握手成功的问题，当服务端连续N次心跳超时之后需要主动关闭链路，清空该客户端的地址缓存信息，以保证后续该客户端可以重连成功，防止被重复登录保护机制拒绝掉。

4.消息缓存重发

无论客户端还是服务端，当发生链路中断之后，在链路恢复之前，缓存在消息队列中待发送的消息不能丢失，等链路恢复之后，重新发送这些消息，保证链路中断期间消息不丢失。

考虑到内存溢出的风险，建议消息缓存队列设置上限，当达到上限之后，应该拒绝继续向该队列添加新的消息。

### 安全性设计 ###

为了保证整个集群环境的安全，内部长连接采用基于IP地址的安全认证机制，服务端对握手请求消息的IP地址进行合法性校验：如果在白名单之内，则校验通过；否则，拒绝对方连接。

如果将Netty协议栈放到公网中使用，需要采用更加严格的安全认证机制，例如基于密钥和AES加密的用户名+密码认证机制，也可以采用SSL/TSL安全传输。

作为示例程序，Netty协议栈采用最简单的基于IP地址的白名单安全认证机制。

### 可扩展性设计 ###

Netty协议需要具备一定的扩展能力，业务可以在消息头中自定义业务域字段，例如消息流水号、业务自定义消息头等。通过Netty消息头中的可选附件attachment字段，业务可以方便地进行自定义扩展。

Netty协议栈架构需要具备一定的扩展能力，例如统一的消息拦截、接口日志、安全、加解密等可以被方便地添加和删除，不需要修改之前的逻辑代码，类似Servlet的FilterChain和AOP，但考虑到性能因素，不推荐通过AOP来实现功能的扩展。

## Netty协议栈开发 ##

### 数据结构定义  ###

![复制代码][copycode.gif]

import lombok.Data;
    
    @Data
    public final class NettyMessage {
        private Header header; //消息头
        private Object body;//消息体
    
        @Override
        public String toString() {
            return "NettyMessage [header=" + header + "]";
        }
    }
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Data
    public final class Header {
        private int crcCode = 0xabef0101;
        private int length;// 消息长度
        private long sessionID;// 会话ID
        private byte type;// 消息类型
        private byte priority;// 消息优先级
        private Map attachment = new HashMap(); // 附件
    
        @Override
        public String toString() {
            return "Header [crcCode=" + crcCode + ", length=" + length
                    + ", sessionID=" + sessionID + ", type=" + type + ", priority="
                    + priority + ", attachment=" + attachment + "]";
        }
    }

![复制代码][copycode.gif]

### 消息编解码 ###

![复制代码][copycode.gif]

import io.netty.buffer.ByteBuf;
    import org.jboss.marshalling.*;
    
    import java.io.IOException;
    
    public class MarshallingDecoder {
        private final Unmarshaller unmarshaller;
    
        public MarshallingDecoder() throws IOException {
            final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
            final MarshallingConfiguration configuration = new MarshallingConfiguration();
            configuration.setVersion(5);
            unmarshaller = marshallerFactory.createUnmarshaller(configuration);
        }
    
        protected Object decode(ByteBuf in) throws Exception {
            int objectSize = in.readInt();
            ByteBuf buf = in.slice(in.readerIndex(), objectSize);
            ByteInput input = new ChannelBufferByteInput(buf);
            try {
                unmarshaller.start(input);
                Object obj = unmarshaller.readObject();
                unmarshaller.finish();
                in.readerIndex(in.readerIndex() + objectSize);
                return obj;
            } finally {
                unmarshaller.close();
            }
        }
    }
    import io.netty.buffer.ByteBuf;
    import org.jboss.marshalling.Marshaller;
    import org.jboss.marshalling.MarshallerFactory;
    import org.jboss.marshalling.Marshalling;
    import org.jboss.marshalling.MarshallingConfiguration;
    
    import java.io.IOException;
    
    public class MarshallingEncoder {
        private static final byte[] LENGTH_PLACEHOLDER = new byte[4];
        Marshaller marshaller;
    
        public MarshallingEncoder() throws IOException {
            final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
            final MarshallingConfiguration configuration = new MarshallingConfiguration();
            configuration.setVersion(5);
            marshaller = marshallerFactory.createMarshaller(configuration);
        }
    
        protected void encode(Object msg, ByteBuf out) throws Exception {
            try {
                int lengthPos = out.writerIndex();
                out.writeBytes(LENGTH_PLACEHOLDER);
                ChannelBufferByteOutput output = new ChannelBufferByteOutput(out);
                marshaller.start(output);
                marshaller.writeObject(msg);
                marshaller.finish();
                out.setInt(lengthPos, out.writerIndex() - lengthPos - 4);
            } finally {
                marshaller.close();
            }
        }
    }
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    
    import java.io.IOException;
    import java.util.HashMap;
    import java.util.Map;
    
    //Netty的LengthFieldBasedFrameDecoder解码器，它支持自动的TCP粘包和半包处理，
    //只需要给出标识消息长度的字段偏移量和消息长度自身所占的字节数，Netty就能自动实现对半包的处理。
    public class NettyMessageDecoder extends LengthFieldBasedFrameDecoder {
    
        MarshallingDecoder marshallingDecoder;
    
        public NettyMessageDecoder(int maxFrameLength, int lengthFieldOffset,int lengthFieldLength) throws IOException {
            super(maxFrameLength, lengthFieldOffset, lengthFieldLength,-8,0);
            marshallingDecoder = new MarshallingDecoder();
        }
    
        @Override
        protected Object decode(ChannelHandlerContext ctx, ByteBuf in)throws Exception {
            //对于业务解码器来说，调用父类LengthFieldBasedFrameDecoder的解码方法后，返回的就是整包消息或者为空，
            //如果为空说明是个半包消息，直接返回继续由I/O线程读取后续的码流
            ByteBuf frame = (ByteBuf) super.decode(ctx, in);
            if (frame == null) {
                return null;
            }
            int pre = in.readerIndex();
            in.readerIndex(0);
            NettyMessage message = new NettyMessage();
            Header header = new Header();
            header.setCrcCode(in.readInt());
            header.setLength(in.readInt());
            header.setSessionID(in.readLong());
            header.setType(in.readByte());
            header.setPriority(in.readByte());
    
            int size = in.readInt();
            if (size > 0) {
                Map<String,Object> attch = new HashMap<String,Object>(size);
                int keySize = 0;
                byte[] keyArray = null;
                String key = null;
                for (int i = 0; i < size; i++) {
                    keySize = in.readInt();
                    keyArray = new byte[keySize];
                    in.readBytes(keyArray);
                    key = new String(keyArray, "UTF-8");
                    attch.put(key, marshallingDecoder.decode(in));
                }
                keyArray = null;
                key = null;
                header.setAttachment(attch);
            }
            if (in.readableBytes() > 4) {
                message.setBody(marshallingDecoder.decode(in));
            }
            in.readerIndex(pre);
            message.setHeader(header);
            return message;
        }
    }
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToMessageEncoder;
    
    import java.io.IOException;
    import java.util.List;
    import java.util.Map;
    
    public final class NettyMessageEncoder extends MessageToMessageEncoder {
    
        MarshallingEncoder marshallingEncoder;
    
        public NettyMessageEncoder() throws IOException {
            this.marshallingEncoder = new MarshallingEncoder();
        }
    
        @Override
        protected void encode(ChannelHandlerContext ctx, Object o, List out) throws Exception {
            NettyMessage msg = (NettyMessage) o;
            if (msg == null || msg.getHeader() == null) {
                throw new Exception("The encode message is null");
            }
            ByteBuf sendBuf = Unpooled.buffer();
            sendBuf.writeInt((msg.getHeader().getCrcCode()));
            sendBuf.writeInt((msg.getHeader().getLength()));
            sendBuf.writeLong((msg.getHeader().getSessionID()));
            sendBuf.writeByte((msg.getHeader().getType()));
            sendBuf.writeByte((msg.getHeader().getPriority()));
            sendBuf.writeInt((msg.getHeader().getAttachment().size()));
            String key = null;
            byte[] keyArray = null;
            Object value = null;
            for (Map.Entry param : msg.getHeader().getAttachment().entrySet()) {
                key = (String) param.getKey();
                keyArray = key.getBytes("UTF-8");
                sendBuf.writeInt(keyArray.length);
                sendBuf.writeBytes(keyArray);
                value = param.getValue();
                marshallingEncoder.encode(value, sendBuf);
            }
            key = null;
            keyArray = null;
            value = null;
            if (msg.getBody() != null) {
                marshallingEncoder.encode(msg.getBody(), sendBuf);
            } else {
                sendBuf.writeInt(0);
            }
            sendBuf.setInt(4, sendBuf.readableBytes());
    
            out.add(sendBuf);
        }
    }

![复制代码][copycode.gif]

### 握手和安全认证  ###

![复制代码][copycode.gif]

import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class LoginAuthReqHandler extends ChannelHandlerAdapter {
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            //当客户端跟服务端TCP三次握手成功之后，由客户端构造握手请求消息发送给服务端
            ctx.writeAndFlush(buildLoginReq());
        }
    
        // 握手请求发送之后，按照协议规范，服务端需要返回握手应答消息。
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {
            NettyMessage message = (NettyMessage) msg;
    
            // 如果是握手应答消息，需要判断是否认证成功
            //对握手应答消息进行处理，首先判断消息是否是握手应答消息，
            if (message.getHeader() != null &&
                    message.getHeader().getType() == MessageType.LOGIN_RESP.value()) {
                byte loginResult = (Byte) message.getBody();
                if (loginResult != (byte) 0) {
                    // 如果是握手应答消息，则对应答结果进行判断，如果非0，说明认证失败，关闭链路，重新发起连接。
                    // 握手失败，关闭连接
                    ctx.close();
                } else {
                    System.out.println("Login is ok : " + message);
                    ctx.fireChannelRead(msg);
                }
            } else {
                // 如果不是，直接透传给后面的ChannelHandler进行处理；
                ctx.fireChannelRead(msg);
            }
        }
    
        private NettyMessage buildLoginReq() {
            // 由于采用IP白名单认证机制，因此，不需要携带消息体，消息体为空，消息类型为3：握手请求消息。
            NettyMessage message = new NettyMessage();
            Header header = new Header();
            header.setType(MessageType.LOGIN_REQ.value());
            message.setHeader(header);
            return message;
        }
    
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.fireExceptionCaught(cause);
        }
    }
    
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    import java.net.InetSocketAddress;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    public class LoginAuthRespHandler extends ChannelHandlerAdapter {
    
        private Map nodeCheck = new ConcurrentHashMap();
        private String[] whitekList = {"127.0.0.1","10.100.1.122"};
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {
            NettyMessage message = (NettyMessage) msg;
    
            // 如果是握手请求消息，处理，其他消息透传
            if (message.getHeader() != null
                    && message.getHeader().getType() == MessageType.LOGIN_REQ.value()) {
    
                String nodeIndex = ctx.channel().remoteAddress().toString();
                NettyMessage loginResp = null;
                // 重复登录，拒绝
                // 重复登录保护
                if (nodeCheck.containsKey(nodeIndex)) {
                    loginResp = buildResponse((byte) -1);
                } else {
                    //IP认证白名单列表
                    InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
                    String ip = address.getAddress().getHostAddress();
                    boolean isOK = false;
                    for (String WIP : whitekList) {
                        if (WIP.equals(ip)) {
                            isOK = true;
                            break;
                        }
                    }
                    //通过buildResponse构造握手应答消息返回给客户端
                    loginResp = isOK ? buildResponse((byte) 0) : buildResponse((byte) -1);
                    if (isOK) {
                        nodeCheck.put(nodeIndex, true);
                    }
                }
                System.out.println("The login response is : " + loginResp + " body [" + loginResp.getBody() + "]");
                ctx.writeAndFlush(loginResp);
            } else {
                ctx.fireChannelRead(msg);
            }
        }
    
        private NettyMessage buildResponse(byte result) {
            NettyMessage message = new NettyMessage();
            Header header = new Header();
            header.setType(MessageType.LOGIN_RESP.value());
            message.setHeader(header);
            message.setBody(result);
            return message;
        }
    
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {
            //当发生异常关闭链路的时候，需要将客户端的信息从登录注册表中去注册，以保证后续客户端可以重连成功。
            nodeCheck.remove(ctx.channel().remoteAddress().toString());//删除缓存
            ctx.close();
            ctx.fireExceptionCaught(cause);
        }
    }

![复制代码][copycode.gif]

### 心跳检测机制 ###

![复制代码][copycode.gif]

import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.util.concurrent.ScheduledFuture;
    
    import java.util.concurrent.TimeUnit;
    
    public class HeartBeatReqHandler extends ChannelHandlerAdapter {
    
        private volatile ScheduledFuture heartBeat;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            NettyMessage message = (NettyMessage) msg;
            // 握手成功，主动发送心跳消息
            //HeartBeatReqHandler接收到之后对消息进行判断
            if (message.getHeader() != null
                    && message.getHeader().getType() == MessageType.LOGIN_RESP.value()) {
                //当握手成功之后，握手请求Handler会继续将握手成功消息向下透传
                //如果是握手成功消息，则启动无限循环定时器用于定期发送心跳消息。
                //由于NioEventLoop是一个schedule，因此它支持定时器的执行。
                // 心跳定时器的单位是毫秒，默认为5000，即每5秒发送一条心跳消息。
                heartBeat = ctx.executor().scheduleAtFixedRate(
                        new HeartBeatReqHandler.HeartBeatTask(ctx), 0, 5000,
                        TimeUnit.MILLISECONDS);
            } else if (message.getHeader() != null
                    && message.getHeader().getType() == MessageType.HEARTBEAT_RESP.value()) {
                //接收服务端发送的心跳应答消息，并打印客户端接收和发送的心跳消息。
                System.out.println("Client receive server heart beat message : ---> "+ message);
            } else {
                //当握手成功之后，握手请求Handler会继续将握手成功消息向下透传
                ctx.fireChannelRead(msg);
            }
        }
    
        private class HeartBeatTask implements Runnable {
            private final ChannelHandlerContext ctx;
    
            public HeartBeatTask(final ChannelHandlerContext ctx) {
                this.ctx = ctx;
            }
    
            @Override
            public void run() {
                NettyMessage heatBeat = buildHeatBeat();
                System.out.println("Client send heart beat messsage to server : ---> "+ heatBeat);
                ctx.writeAndFlush(heatBeat);
            }
    
            private NettyMessage buildHeatBeat() {
                NettyMessage message = new NettyMessage();
                Header header = new Header();
                header.setType(MessageType.HEARTBEAT_REQ.value());
                message.setHeader(header);
                return message;
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            if (heartBeat != null) {
                heartBeat.cancel(true);
                heartBeat = null;
            }
            ctx.fireExceptionCaught(cause);
        }
    }
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class HeartBeatRespHandler extends ChannelHandlerAdapter {
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {
            NettyMessage message = (NettyMessage) msg;
            // 返回心跳应答消息
            if (message.getHeader() != null
                    && message.getHeader().getType() == MessageType.HEARTBEAT_REQ.value()) {
                System.out.println("Receive client heart beat message : ---> "+ message);
                NettyMessage heartBeat = buildHeatBeat();
                System.out.println("Send heart beat response message to client : ---> "+ heartBeat);
                ctx.writeAndFlush(heartBeat);
            } else {
                ctx.fireChannelRead(msg);
            }
        }
    
        private NettyMessage buildHeatBeat() {
            NettyMessage message = new NettyMessage();
            Header header = new Header();
            header.setType(MessageType.HEARTBEAT_RESP.value());
            message.setHeader(header);
            return message;
        }
    }

![复制代码][copycode.gif]

### 基础类 ###

![复制代码][copycode.gif]

public enum MessageType {
    
        LOGIN_REQ((byte)3),
        LOGIN_RESP((byte)4),
        HEARTBEAT_REQ((byte)5),
        HEARTBEAT_RESP((byte)6),
    
        ;
    
        public byte value;
    
        MessageType(byte v){
            this.value = v;
        }
    
        public byte value(){
            return value;
        }
    
    }
    import io.netty.buffer.ByteBuf;
    import org.jboss.marshalling.ByteInput;
    
    import java.io.IOException;
    
    public class ChannelBufferByteInput implements ByteInput {
    
        private final ByteBuf buffer;
    
        ChannelBufferByteInput(ByteBuf buffer) {
            this.buffer = buffer;
        }
    
        @Override
        public void close() throws IOException {
        }
    
        @Override
        public int available() throws IOException {
            return buffer.readableBytes();
        }
    
        @Override
        public int read() throws IOException {
            if (buffer.isReadable()) {
                return buffer.readByte() & 0xff;
            }
            return -1;
        }
    
        @Override
        public int read(byte[] array) throws IOException {
            return read(array, 0, array.length);
        }
    
        @Override
        public int read(byte[] dst, int dstIndex, int length) throws IOException {
            int available = available();
            if (available == 0) {
                return -1;
            }
    
            length = Math.min(available, length);
            buffer.readBytes(dst, dstIndex, length);
            return length;
        }
    
        @Override
        public long skip(long bytes) throws IOException {
            int readable = buffer.readableBytes();
            if (readable < bytes) {
                bytes = readable;
            }
            buffer.readerIndex((int) (buffer.readerIndex() + bytes));
            return bytes;
        }
    
    }
    import io.netty.buffer.ByteBuf;
    import org.jboss.marshalling.ByteOutput;
    
    import java.io.IOException;
    
    public class ChannelBufferByteOutput implements ByteOutput {
    
        private final ByteBuf buffer;
    
        ChannelBufferByteOutput(ByteBuf buffer) {
            this.buffer = buffer;
        }
    
        @Override
        public void close() throws IOException {
            // Nothing to do
        }
    
        @Override
        public void flush() throws IOException {
            // nothing to do
        }
    
        @Override
        public void write(int b) throws IOException {
            buffer.writeByte(b);
        }
    
        @Override
        public void write(byte[] bytes) throws IOException {
            buffer.writeBytes(bytes);
        }
    
        @Override
        public void write(byte[] bytes, int srcIndex, int length) throws IOException {
            buffer.writeBytes(bytes, srcIndex, length);
        }
    
        /**
         * Return the {
        @link ByteBuf} which contains the written content
         *
         */
        ByteBuf getBuffer() {
            return buffer;
        }
    }import io.netty.buffer.ByteBuf;
    import org.jboss.marshalling.ByteOutput;
    
    import java.io.IOException;
    
    public class ChannelBufferByteOutput implements ByteOutput {
    
        private final ByteBuf buffer;
    
        ChannelBufferByteOutput(ByteBuf buffer) {
            this.buffer = buffer;
        }
    
        @Override
        public void close() throws IOException {
            // Nothing to do
        }
    
        @Override
        public void flush() throws IOException {
            // nothing to do
        }
    
        @Override
        public void write(int b) throws IOException {
            buffer.writeByte(b);
        }
    
        @Override
        public void write(byte[] bytes) throws IOException {
            buffer.writeBytes(bytes);
        }
    
        @Override
        public void write(byte[] bytes, int srcIndex, int length) throws IOException {
            buffer.writeBytes(bytes, srcIndex, length);
        }
    
        /**
         * Return the {
        @link ByteBuf} which contains the written content
         *
         */
        ByteBuf getBuffer() {
            return buffer;
        }
    }

![复制代码][copycode.gif]

### 客户端 ###

![复制代码][copycode.gif]

import io.netty.buffer.ByteBuf;
    import org.jboss.marshalling.ByteOutput;
    
    import java.io.IOException;
    
    public class ChannelBufferByteOutput implements ByteOutput {
    
        private final ByteBuf buffer;
    
        ChannelBufferByteOutput(ByteBuf buffer) {
            this.buffer = buffer;
        }
    
        @Override
        public void close() throws IOException {
            // Nothing to do
        }
    
        @Override
        public void flush() throws IOException {
            // nothing to do
        }
    
        @Override
        public void write(int b) throws IOException {
            buffer.writeByte(b);
        }
    
        @Override
        public void write(byte[] bytes) throws IOException {
            buffer.writeBytes(bytes);
        }
    
        @Override
        public void write(byte[] bytes, int srcIndex, int length) throws IOException {
            buffer.writeBytes(bytes, srcIndex, length);
        }
    
        /**
         * Return the {
        @link ByteBuf} which contains the written content
         *
         */
        ByteBuf getBuffer() {
            return buffer;
        }
    }
    
    public class NettyConstant {
    
        public static String LOCALIP = "127.0.0.1";
        public static String REMOTEIP = "127.0.0.1";
        public static Integer LOCAL_PORT = 8085;
        public static Integer PORT = 9099;
    }

![复制代码][copycode.gif]

### 服务端： ###

![复制代码][copycode.gif]

import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    import io.netty.handler.timeout.ReadTimeoutHandler;
    
    import java.io.IOException;
    
    public class NettyServer {
    
        public void bind() throws Exception {
            // 配置服务端的NIO线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 100)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer() {
                        @Override
                        public void initChannel(Channel ch)throws IOException{
                            ch.pipeline().addLast(new NettyMessageDecoder(1024 * 1024, 4, 4));
                            ch.pipeline().addLast(new NettyMessageEncoder());
                            ch.pipeline().addLast("readTimeoutHandler",new ReadTimeoutHandler(50));
                            ch.pipeline().addLast(new LoginAuthRespHandler());
                            ch.pipeline().addLast("HeartBeatHandler",new HeartBeatRespHandler());
                        }
                    });
    
            // 绑定端口，同步等待成功
            b.bind(NettyConstant.REMOTEIP, NettyConstant.PORT).sync();
            System.out.println("Netty server start ok : " + (NettyConstant.REMOTEIP + " : " + NettyConstant.PORT));
        }
    
        public static void main(String[] args) throws Exception {
            new NettyServer().bind();
        }
    }

![复制代码][copycode.gif]

### 测试结果： ###

服务端

![复制代码][copycode.gif]

12:30:32.998 [nioEventLoopGroup-2-1] INFO  i.n.handler.logging.LoggingHandler - [id: 0x4d893ed4, /127.0.0.1:9099] RECEIVED: [id: 0x343516a3, /127.0.0.1:8085 => /127.0.0.1:9099]
    12:30:33.205 [nioEventLoopGroup-3-1] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.leakDetectionLevel: simple
    The login response is : NettyMessage [header=Header [crcCode=-1410399999, length=0, sessionID=0, type=4, priority=0, attachment={}]] body [0]
    Receive client heart beat message : ---> NettyMessage [header=Header [crcCode=-1410399999, length=26, sessionID=0, type=5, priority=0, attachment={}]]
    Send heart beat response message to client : ---> NettyMessage [header=Header [crcCode=-1410399999, length=0, sessionID=0, type=6, priority=0, attachment={}]]
    Receive client heart beat message : ---> NettyMessage [header=Header [crcCode=-1410399999, length=26, sessionID=0, type=5, priority=0, attachment={}]]
    Send heart beat response message to client : ---> NettyMessage [header=Header [crcCode=-1410399999, length=0, sessionID=0, type=6, priority=0, attachment={}]]
    Receive client heart beat message : ---> NettyMessage [header=Header [crcCode=-1410399999, length=26, sessionID=0, type=5, priority=0, attachment={}]]
    Send heart beat response message to client : ---> NettyMessage [header=Header [crcCode=-1410399999, length=0, sessionID=0, type=6, priority=0, attachment={}]]

![复制代码][copycode.gif]

客户端

![复制代码][copycode.gif]

12:30:33.152 [nioEventLoopGroup-2-1] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.leakDetectionLevel: simple
    Login is ok : NettyMessage [header=Header [crcCode=-1410399999, length=101, sessionID=0, type=4, priority=0, attachment={}]]
    Client send heart beat messsage to server : ---> NettyMessage [header=Header [crcCode=-1410399999, length=0, sessionID=0, type=5, priority=0, attachment={}]]
    Client receive server heart beat message : ---> NettyMessage [header=Header [crcCode=-1410399999, length=26, sessionID=0, type=6, priority=0, attachment={}]]
    Client send heart beat messsage to server : ---> NettyMessage [header=Header [crcCode=-1410399999, length=0, sessionID=0, type=5, priority=0, attachment={}]]
    Client receive server heart beat message : ---> NettyMessage [header=Header [crcCode=-1410399999, length=26, sessionID=0, type=6, priority=0, attachment={}]]
    Client send heart beat messsage to server : ---> NettyMessage [header=Header [crcCode=-1410399999, length=0, sessionID=0, type=5, priority=0, attachment={}]]
    Client receive server heart beat message : ---> NettyMessage [header=Header [crcCode=-1410399999, length=26, sessionID=0, type=6, priority=0, attachment={}]]
    Client send heart beat messsage to server : ---> NettyMessage [header=Header [crcCode=-1410399999, length=0, sessionID=0, type=5, priority=0, attachment={}]]

![复制代码][copycode.gif]

[990532-20161215161242683-1070510244.png]: /images/20220612/56e11f14bad544cb9e00764237328edc.png
[990532-20161215161437995-1502340572.png]: /images/20220612/58b99e553e4e4d6abacdb388d90b9445.png
[990532-20161215161904339-1672827031.png]: /images/20220612/e3584c74c47048ba8a88990fbb39fe44.png
[990532-20161215161915479-351556821.png]: /images/20220612/b608d4a389e54a6c9109452faa9a3f58.png
[990532-20161215161937261-1103661657.png]: /images/20220612/d00026a1a2d345ae926532ea6242a07b.png
[990532-20161215162023745-539949159.png]: /images/20220612/c3f16a824e404cbfb0fec789e8f57d15.png
[copycode.gif]: /images/20220612/44636a7cfcb44763bbb6337e6302b0fd.png