Netty网编程实战：四种解决粘包方式切换、两种生产级双向监听模式并行、高效编解码、多处理器协同作战-低调大师

Netty网编程实战：四种解决粘包方式切换、两种生产级双向监听模式并行、高效编解码、多处理器协同作战

2020-07-01 529

前言

在前面的课题把Netty网络编程的各种理论和主要组件、核心部件、重要机制都摊开说清楚后，今天给大家分享一下Netty在网络编程中的一些典型应用场景和一些重要问题的解决方案。比如由于TCP网络传输底层的原因，而产生不可预期的粘包和半包问题，致使收到对端的报文明显缺斤少两或前后报文之间相互粘连在一起；又如在生产环境中，网络通信的任意一端由于外界网络或自身bug的原因，导致网络长时间阻塞或直接断开等行为，对端无法明显感知真正原因；不能第一时间通知到对端或相关技术人员处理，最终导致更大的业务受损。

网络粘包半包

假设客户端分别发送了两个数据包 D1 和 D2 给服务端，由于服务端一次读取到的字节数是不确定的，故可能存在以下 4 种情况。

（1）服务端分两次读取到了两个独立的数据包，分别是 D1 和 D2，没有粘包和拆包；

（2）服务端一次接收到了两个数据包，D1 和 D2 粘合在一起，被称为 TCP 粘包；

（3）服务端分两次读取到了两个数据包，第一次读取到了完整的 D1 包和 D2 包的部分内容，第二次读取到了 D2 包的剩余内容，这被称为 TCP 拆包；

（4）服务端分两次读取到了两个数据包，第一次读取到了 D1 包的部分内容 D1_1，第二次读取到了 D1 包的剩余内容 D1_2 和 D2 包的整包。

如果此时服务端 TCP 接收滑窗非常小，而数据包 D1 和 D2 比较大，很有可能会发生第五种可能，即服务端分多次才能将 D1 和 D2 包接收完全，期间发生多次拆包。

由于 TCP 协议本身的机制（面向连接的可靠地协议-三次握手机制）客户端与服务器会维持一个连接，数据在连接不断开的情况下，可以持续不断地将多个数据包发往服务器，但是如果发送的网络数据包太小，那么他本身会启用 Nagle 算法（可配置是否启用）对较小的数据包进行合并然后再发送。那么这样的话，服务器在接收到消息（数据流）的时候就无法区分哪些数据包是客户端自己分开发送的，这样产生了粘包；服务器在接收到数据库后，放到缓冲区中，如果消息没有被及时从缓存区取走，下次在取数据的时候可能就会出现一次取出多个数据包的情况，造成粘包现象。

解决办法已在前面的文章介绍得很清楚了，这里不在赘述。今天本文将综合采用回车字符分割、任意特殊字符分割、固定字符分割和数据本身带长度信息等四种方式，交替切换，在实战层面给大家展示这种四种方式是如何构建和工作的。

生产级双向心跳机制

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

本文将提供双向端对端心跳检测机制滑动检测，随时把双向服务的运行动向级存活情况，体现在前端控制台。

生产级存活检测和自动滑动重试与恢复机制

自动监听对端通道是否断开或突变为未激活状态，致使网络传输中断或长时间阻塞等待。当这些问题发生时，对端会在第一时间感知到并启动自动滑动重试(重试频率在持续断开时，将逐渐降低)机制，当对端网络或服务恢复时，当前端也自动恢复正常的通讯，不需要外接的任何干预。

高效编解码器

以上是Facebook在高连接数情况下，综合对以上20种三方开源序列和反序列化组件，进行对比测试。大家都知道JDK内置的基于流的序列化和反序列化效率都非常低下，数据报文和原文相比也膨胀得相当庞大。本文将以google的原生protostuff组件为例，展示Netty内嵌非内置组件的方式和高效轻快的序列化表现。

各种ChannelHandler协同

在ChannelPipeline内部维护一个双向链表来存储进入管道的，各种ChannelHandler事件和处理器有序责任链式处理有业务执行先后顺序的处理及事件触发、传递等。倘若我们打乱Pipeline中处理器的顺序，那么执行最终会报错终止或数据报文传递和处理结果错乱。本文将容器化载入多种Handler实现多业务处理协同工作。

有了上面本文核心功能和业务集成介绍后，下面我们就运用Netty强大的特性，完成我们需集成的所有功能的代码设计。

各部分代码实战

一、客户端部分

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：Netty客户端业务通道处理器类
 * 负责处理和服务端的所有IO业务事件处理
 */
public class ClientChannelHandler extends SimpleChannelInboundHandler<ByteBuf> {
    /**
     * 统计当前通道第几次读取数据
     */
    private final AtomicInteger counter = new AtomicInteger(0);
    /**
     * 客户端读取网络通道数据后处理
     * @param channelHandlerContext 通道上下文(很重要,网络IO业务就靠它了)
     * @param byteBuf 网络传送过来的数据
     * @throws Exception
     */
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
        //打印接收到的服务端数据和当前统计计数
        System.out.println("Accept server data {"+byteBuf.toString(CharsetUtil.UTF_8)+"], the counter is:"+this.counter.incrementAndGet());
    }

    /**
     * 连接建立成功事件回调
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //连接建立后，循环10次向服务端发送连续报文
        ByteBuf buf = null;
        String msg = null;
        for (int i=0;i<10;i++){
            if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.SEND_LEN){
                UserAddress address = new UserAddress("abc@163.com", "ChengDu.SiChuan");
                User user = new User(i,"AndyChen"+i,"WaveBeed"+i, address);
                ctx.write(user);
            }else{
                msg = "Client message"+i+" data ";
                if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.LINE_BASED){
                    msg += "(line_based)"+System.getProperty("line.separator");
                }
                else if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.DELIMITER_BASED){
                    msg += "(custom_based)"+Constant.CUSTOM_SPLIT_MARK;
                }else if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.FIXED_LEN){
                    msg = Constant.FIXED_LEN_CLIENT_TXT;
                }
                buf = Unpooled.buffer(msg.length());
                buf.writeBytes(msg.getBytes());
                ctx.writeAndFlush(buf);
            }
        }
        if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.SEND_LEN){
            ctx.flush();
        }
    }
    /**
     * 处理器异常集中处理
     * @param ctx 处理器上下文
     * @param cause 异常
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         cause.printStackTrace();
         ctx.close();
    }
}

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：Netty客户端启动器类
 */
public class NettyClientStarter {
    //采用netty内置定时器
    private static final HashedWheelTimer timer = new HashedWheelTimer();
    /**
     * 启动客户端
     * @param args
     */
    public static void main(String[] args){
        try {
            start();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 启动客户端
     */
    private static void start() throws InterruptedException {
        //线程组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //Netty客户端启动类
            final Bootstrap boot = new Bootstrap();
            boot.group(group)//将线程组绑定到启动器
                    .channel(NioSocketChannel.class)//使用NIO进行网络传输
                    //绑定服务端连接地址
                    .remoteAddress(new InetSocketAddress(Constant.SERV_HOST, Constant.SERV_PORT));
            /**
             * 定义监听器
             */
            final ChannelMonitor<Channel> monitor = new ClientMonitorHandler(Constant.SERV_HOST, Constant.SERV_PORT, timer, boot);
            boot.handler(new ChannelInitializer<Channel>(){
                protected void initChannel(Channel channel) throws Exception {
                    channel.pipeline().addLast(monitor.setMonitorHandlers());
                }
            });
            //这里会阻塞，直到连接完成
            ChannelFuture future = boot.connect().sync();
            System.out.println("Has connected to server："+Constant.SERV_PORT+" ...");
            //这里也会阻塞，直到连接通道关闭
            future.channel().closeFuture().sync();
        }finally {
            //线程池组关闭
            //group.shutdownGracefully().sync();
        }
    }
}

二、服务端部分

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：Netty服务端业务通道处理器类
 * 负责向对端响应业务应答和服务端相关IO业务处理
 */
public class ServerChannelHandler extends ChannelInboundHandlerAdapter {
    /**
     * 消息接收计数器
     */
    private final AtomicInteger counter = new AtomicInteger(0);
    /**
     * 接收客户端发送的数据
     * @param ctx 处理器上下文
     * @param msg 消息
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String clientData = null;
        User user = null;
        if(msg instanceof User){
            user = (User)msg;
            clientData = user.toString();
        }else{
            clientData = ((ByteBuf)msg).toString(CharsetUtil.UTF_8);
        }
        System.out.println("Accept client data ["+clientData+"], the counter is:"+this.counter.incrementAndGet());
        //回馈消息给客户端
        String toClientData = "Data has been accepted by server";
        if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.LINE_BASED ||
           Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.SEND_LEN){
            toClientData += "(line_based)"+System.getProperty("line.separator");
        }
        else if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.DELIMITER_BASED){
            toClientData += "(custom_based)"+Constant.CUSTOM_SPLIT_MARK;
        }else if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.FIXED_LEN){
            toClientData = Constant.FIXED_LEN_SERVER_TXT;
        }
        ctx.writeAndFlush(Unpooled.copiedBuffer(toClientData.getBytes()));
        if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.SEND_LEN && null != user){
            ctx.fireChannelRead(user);
        }
    }
    /**
     *
     * @param ctx 处理器上下文
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelReadComplete();
        System.out.println("Client data recevied completed!");
    }

    /**
     * 管道关闭时触发
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Client: "+ctx.channel().remoteAddress()+" channel will close...");
    }
    /**
     * 异常处理器
     * @param ctx 处理器上下文
     * @param cause 异常
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：Netty服务端启动器类
 */
public class NettyServerStarter {
    /**
     * 启动服务器
     * @param args
     */
    public static void main(String[] args) {
        try {
            start();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 启动服务器端
     */
    private static void start() throws InterruptedException {
        //线程组
        EventLoopGroup mainGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            //服务端启动器
            ServerBootstrap boot = new ServerBootstrap();
            boot.group(mainGroup, workGroup)//将线程组绑定到启动器
                    .channel(NioServerSocketChannel.class)//使用NIO进行网络通讯
                    .localAddress(new InetSocketAddress(Constant.SERV_PORT))//绑定本地端口监听
                    .childHandler(new ChannelInitializerExt());//为Channel添加业务处理器
            //异步绑定服务器端口，sync()方法阻塞，直到绑定完成
            ChannelFuture future = boot.bind().sync();
            System.out.println("Server address:"+Constant.SERV_PORT+" has bind complete，waiting for data...");
            //这里通道会阻塞，直到通道关闭
            future.channel().closeFuture().sync();
        }finally {
            //优雅地关闭线程池组
            mainGroup.shutdownGracefully().sync();
            workGroup.shutdownGracefully().sync();
        }
    }
    /**
     * 通道初始化器扩展
     * 负责定义初始化各种业务ChannelHandler
     */
    private static class ChannelInitializerExt extends ChannelInitializer<Channel>{
        /**
         * 初始化Channel
         * @param channel 当前通道
         * @throws Exception
         */
        protected void initChannel(Channel channel) throws Exception {
            ChannelPipeline p = channel.pipeline();
            //添加对客户端心跳检测
            p.addLast("heartbeat-state",new IdleStateHandler(Constant.HEART_BEAT_TIMEOUT,0,0, TimeUnit.SECONDS));
            p.addLast("heartbeat-check", new ServerHeartbeatHandler());
            /**
             * 解决粘包半包问题办法一：字符分隔
             */
            //支持一：回车分隔
            if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.LINE_BASED){
                p.addLast("linebase", new LineBasedFrameDecoder(Constant.DECODER_BYTE_MAX_LEN));
            }
            //支持二：自定义分隔符分隔
            if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.DELIMITER_BASED) {
                ByteBuf byteMark = Unpooled.copiedBuffer(Constant.CUSTOM_SPLIT_MARK.getBytes());
                p.addLast("custom", new DelimiterBasedFrameDecoder(Constant.DECODER_BYTE_MAX_LEN, byteMark));
            }
            /**
             * 解决粘包半包问题办法二：固定长度
             */
            if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.FIXED_LEN) {
                p.addLast("fixlength", new FixedLengthFrameDecoder(Constant.FIXED_LEN_CLIENT_TXT.length()));
            }
            /**
             * 解决粘包半包问题办法三：带长度
             */
            if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.SEND_LEN) {
                p.addLast("sendlen", new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
                p.addLast(new ProtoStuffDecoder<User>(User.class));
                p.addLast("linebase", new LineBasedFrameDecoder(Constant.DECODER_BYTE_MAX_LEN));
            }
            //业务处理Handler--往往此Handler是注册在管道的最后节点
            p.addLast(new ServerChannelHandler());
        }
    }
}

三、心跳机制部分

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：客户端心跳检测处理器
 */
public class ClientHeartbeatHandler extends ChannelInboundHandlerAdapter {
    /**
     * 事件触发器
     * @param ctx 处理器上下文
     * @param evt 触发对象
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        /**
         * 构建心跳字节序列
         */
        String seq = Constant.HEART_BEAT_CLIN_MSG;
        if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.LINE_BASED ||
           Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.SEND_LEN){
            seq += System.getProperty("line.separator");
        }
        else if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.DELIMITER_BASED){
            seq += Constant.CUSTOM_SPLIT_MARK;
        }else if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.FIXED_LEN){
            seq = Constant.FIXED_LEN_SERVER_TXT;
        }
        final ByteBuf HEARBEAT_SEQ = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer(seq, CharsetUtil.UTF_8));//ISO_8859_1
        /**
         * 发送心跳消息，并在发送失败时关闭连接
         */
        if(evt instanceof IdleStateEvent){
            IdleState state = ((IdleStateEvent)evt).state();
            /**
             * 服务端是数据接收端，那么客户端就是Write
             * 客户端5秒钟未收到应答，则认为发送心跳包检测
             */
            if(state == IdleState.WRITER_IDLE){
                ctx.writeAndFlush(HEARBEAT_SEQ).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        }else{
            //非心跳事件，将它传递给下个处理器处理
            super.userEventTriggered(ctx, evt);
        }
    }
}

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：服务端心跳检测处理器
 */
public class ServerHeartbeatHandler extends ChannelInboundHandlerAdapter {
    /**
     * 事件触发器
     * @param ctx 处理器上下文
     * @param evt 触发对象
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        /**
         * 构建心跳字节序列
         */
        String seq = Constant.HEART_BEAT_SERV_MSG;
        if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.LINE_BASED ||
           Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.SEND_LEN){
            seq += System.getProperty("line.separator");
        }
        else if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.DELIMITER_BASED){
            seq += Constant.CUSTOM_SPLIT_MARK;
        }else if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.FIXED_LEN){
            seq = Constant.FIXED_LEN_SERVER_TXT;
        }
        final ByteBuf HEARBEAT_SEQ = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer(seq, CharsetUtil.UTF_8));//ISO_8859_1
        /**
         * 发送心跳消息，并在发送失败时关闭连接
         */
        if(evt instanceof IdleStateEvent){
            IdleState state = ((IdleStateEvent)evt).state();
            /**
             * 客户端是数据发送端，那么服务器端就是Read
             * 服务器端5秒钟未收到应答，则认为发送心跳包检测
             */
            if(state == IdleState.READER_IDLE){
                ctx.writeAndFlush(HEARBEAT_SEQ).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        }else{
            //非心跳事件，将它传递给下个处理器处理
            super.userEventTriggered(ctx, evt);
        }
    }
}

四、生产级检活和重试机制部分

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：通道监视器设计类
 * 负责定时监听与对端的连接，当连接断开时，自动重试连接
 */
@ChannelHandler.Sharable
public abstract class ChannelMonitor<T extends Channel> extends ChannelInboundHandlerAdapter
                                     implements MonitoredChannelHandler, TimerTask {
    private final String host;//监视的主机
    private final int port;//监视的端口
    private final Timer timer;//定时器
    private final AbstractBootstrap boot;//启动对象
    private final AtomicInteger counter = new AtomicInteger(0);//重试次数统计
    public ChannelMonitor(String host, int port, Timer timer, AbstractBootstrap boot) {
        this.host = host;
        this.port = port;
        this.timer = timer;
        this.boot = boot;
    }

    /**
     * 运行定时任务检测通道是否断开
     * @param timeout 定时时间
     * @throws Exception
     */
    public void run(Timeout timeout) throws Exception {
        ChannelFuture future = null;

        /**
         * 设置绑定
         */
        //synchronized (this.boot) {
            if (this.boot instanceof ServerBootstrap) {
                ((ServerBootstrap) this.boot).childHandler(new ChannelInitializerExt());
                future = this.boot.bind(new InetSocketAddress(this.host, this.port));
            } else {
                this.boot.handler(new ChannelInitializerExt());
                future = ((Bootstrap) this.boot).connect(this.host, this.port);
            }
        //}
        /**
         * 监听通道连接
         */
        future.addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                boolean success = channelFuture.isSuccess();
                System.out.println("Retry channel connection "+(success ? "success" : "fail")+"("+counter.get()+"times).");
                if(!success){
                    channelFuture.channel().pipeline().fireChannelInactive();
                }
            }
        });
    }

    /**
     * 连接断开事件处理
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Monitored channel will disconnection...");
        System.out.println("Monitored channel disconnected and will auto reconnect...");
        int times = this.counter.incrementAndGet();
        if(Constant.CH_RETRY_TIMES >= times){
            //逐步延长重试时间
            int timeout = 4 << times;
            this.timer.newTimeout(this, timeout, TimeUnit.MILLISECONDS);
        }
        //触发对下个断开事件的调用
        ctx.fireChannelInactive();
    }
    /**
     * 初始化器扩展
     */
    private class ChannelInitializerExt extends ChannelInitializer<T> {
        //加入handler
        protected void initChannel(T t) throws Exception {
            ChannelHandler[] handlers = setMonitorHandlers();
            t.pipeline().addLast(handlers);
        }
    }
}

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：客户端监视器实现类
 */
@ChannelHandler.Sharable
public class ClientMonitorHandler extends ChannelMonitor {
    public ClientMonitorHandler(String host, int port, Timer timer, AbstractBootstrap boot) {
        super(host, port, timer, boot);
    }

    /**
     * 定义客户端处理器
     * @return
     */
    public ChannelHandler[] setMonitorHandlers() {
        int aLen = (Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.SEND_LEN ? 7 : 5);
        ChannelHandler[] handlers = new ChannelHandler[aLen];
        handlers[0] = this;//监听器本身也是Handler，这里也加入
        handlers[1] = new IdleStateHandler(0, Constant.HEART_BEAT_TIMEOUT+1,0, TimeUnit.SECONDS);
        handlers[2] = new ClientHeartbeatHandler();
        /**
         * 解决粘包半包问题办法一：字符分隔
         */
        //支持一：回车分隔
        if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.LINE_BASED){
            handlers[3] = new LineBasedFrameDecoder(Constant.DECODER_BYTE_MAX_LEN);
        }
        //支持二：自定义分隔符分隔
        if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.DELIMITER_BASED) {
            ByteBuf byteMark = Unpooled.copiedBuffer(Constant.CUSTOM_SPLIT_MARK.getBytes());
            handlers[3] = new DelimiterBasedFrameDecoder(Constant.DECODER_BYTE_MAX_LEN, byteMark);
        }
        /**
         * 解决粘包半包问题办法二：固定长度
         */
        if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.FIXED_LEN) {
            handlers[3] = new FixedLengthFrameDecoder(Constant.FIXED_LEN_SERVER_TXT.length());
        }
        /**
         * 解决粘包半包问题办法三：带长度
         */
        if(Constant.STICKUP_SOLUTION_TYPE == StickupSolutionType.SEND_LEN){
            handlers[3] = new LengthFieldPrepender(2);
            handlers[4] = new ProtoStuffEncoder<User>(User.class);
            handlers[5] = new LineBasedFrameDecoder(Constant.DECODER_BYTE_MAX_LEN);
        }
        handlers[aLen-1] = new ClientChannelHandler();

        return handlers;
    }
}

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：定义被监视的ChannelHandler
 */
public interface MonitoredChannelHandler {
    /**
     * 设置被监视的ChannelHandler
     * @return
     */
    ChannelHandler[] setMonitorHandlers();
}

五、Protostuff编解码器部分

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：ProtoStuff解码器类
 */
public class ProtoStuffDecoder<T> extends MessageToMessageDecoder<ByteBuf> {
    private final RuntimeSchema<T> schema;
    public ProtoStuffDecoder(Class<T> clazz) {
        this.schema = (RuntimeSchema<T>) RuntimeSchema.createFrom(clazz);
    }

    /**
     * 自定义实现高效解码方法
     * @param channelHandlerContext 处理器上下文
     * @param byteBuf 缓冲对象
     * @param list 对象集合
     * @throws Exception
     */
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
        int length = byteBuf.readableBytes();
        if(0 < length){
            byte[] bytes = new byte[length];
            byteBuf.getBytes(byteBuf.readerIndex(), bytes, 0, length);
            T t = this.schema.newMessage();
            ProtostuffIOUtil.mergeFrom(bytes, t, this.schema);
            list.add(t);
        }
    }
}

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：序列化编码器
 */
public class ProtoStuffEncoder<T> extends MessageToByteEncoder<Object> {
    private final RuntimeSchema<T> schema;
    public ProtoStuffEncoder(Class<T> clazz) {
        this.schema = (RuntimeSchema<T>) RuntimeSchema.createFrom(clazz);
    }

    /**
     * 实现高效序列化方法
     * @param channelHandlerContext 处理器上下文
     * @param o 传输对象
     * @param byteBuf buffer缓冲
     * @throws Exception
     */
    protected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) throws Exception {
       byte[] bytes = ProtostuffIOUtil.toByteArray((T)o, this.schema,
               LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE));
       byteBuf.writeBytes(bytes);
    }
}

六、其它协同工具

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：系统相关常量配置
 * 建议直接从全局应用配置文件中
 */
public class Constant {
    /**
     * 服务器端ip
     */
    public static final String SERV_HOST = "127.0.0.1";
    /**
     * 服务端监听端口
     */
    public static final int SERV_PORT = 6888;
    /**
     * 解码器处理字节最大长度
     */
    public static final int DECODER_BYTE_MAX_LEN = 1024;
    /**
     * 固定长度文本
     */
    public static final String FIXED_LEN_CLIENT_TXT = "Netty Client Fixed Len Text";
    /**
     * 固定长度文本
     */
    public static final String FIXED_LEN_SERVER_TXT = "Netty Server Fixed Len Text";
    /**
     * 自定义分隔标记
     */
    public static final String CUSTOM_SPLIT_MARK = "@#";
    /**
     * 粘包解决方法
     */
    public static final StickupSolutionType STICKUP_SOLUTION_TYPE = StickupSolutionType.LINE_BASED;
    /**
     * 心跳超时时间
     */
    public static final int HEART_BEAT_TIMEOUT = 4;
    /**
     * 服务端心跳
     */
    public static final String HEART_BEAT_SERV_MSG = "SERV_HEARTBEAT";
    /**
     * 客户端心跳
     */
    public static final String HEART_BEAT_CLIN_MSG = "CLIN_HEARTBEAT";
    /**
     * 通道重试次数
     */
    public static final int CH_RETRY_TIMES = 30;
}

/**
 * @author andychen https://blog.51cto.com/14815984
 * @description：粘包解决类型定义
 */
public enum StickupSolutionType{
    LINE_BASED(1, "回车分隔"),
    DELIMITER_BASED(2,"自定义分隔"),
    FIXED_LEN(4, "固定长度"),
    SEND_LEN(8, "带长度");


    public int getCode() {
        return code;
    }

    public String getMsg() {
        return msg;
    }

    private final int code;
    private final String msg;
    StickupSolutionType(int code, String msg) {
        this.code = code;
        this.msg = msg;
    }
}

运行结果

总结

基于Netty的网络编程高效性、稳定性、扩展性和易用性远不止于此，后面我们将陆续展示Netty在其它领域的应用和相关实战，给大家分享不同的网络编程场景体验与借鉴。

微信关注我们

原文链接：https://blog.51cto.com/14815984/2507887

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

科学家研制2D金属芯片：让储存速度提高100倍

科学家正在努力，希望开发出下一代数据存储材料，以提高现有存储速度。据英国《自然·物理学》杂志近日发表的一项研究，一个美国联合研究团队利用层状二碲化钨制成了二维（2D）金属芯片，其厚度仅三个原子，其可代替硅芯片存储数据，且比硅芯片更密集、更小、更快，也更节能，同时储存速度提高了 100 倍之多。研究人员对二碲化钨薄层结构施加微小电流，使其奇数层相对于偶数层发生稳定的偏移，并利用奇偶层的排列来存储二进制数据。数据写入后，他们再通过一种称为贝利曲率的量子特性，在不干扰排列的情况下读取数据。与现有的基于硅的数据存储系统相比，新系统具有巨大优势——它可以将更多的数据填充到极小的物理空间中，并且非常节能。此外，其偏移发生得如此之快，以至于数据写入速度可以比现有技术快 100 倍。对超薄层进行非常小的调整，就会对它的功能特性产生很大的影响，而人们可以利用这一知识来设计新型节能设备，以实现可持续发展和更智慧的未来存储方式。

2020-07-03

644

小白写了一堆if-else，大神实在看不下去了，竟然用策略模式直接摆平了

这里涉及到一个关键词：策略模式，那么到底什么是策略模式呢？本文就来好好给大家讲讲策略模式，大家可以带着如下几个问题来阅读本文：如何通过策略模式优化业务逻辑代码（可以根据自己从事的工作思考）使用策略模式优化if-else，会不会带来什么副作用呢？实现策略模式是否有更好的方式呢？ 1. 策略模式如何优化代码解构要会带这个问题，需要先弄清楚策略模式的定义，首先来看策略模式的教科书定义：策略模式(Strategy Pattern)：定义一系列算法，将每一个算法封装起来，并让它们可以相互替换。策略模式让算法独立于使用它的客户而变化，也称为政策模式(Policy)。策略模式是一种对象行为型模式。这里的算法其实就是业务逻辑，为了更形象，干脆将其理解为一个函数。其实这个定义的基本意思就是说，根据策略选择要执行的函数，而每一个策略都会有一个标识，可以称为key。而当前策略要执行的函数，可以称为value，其实就是使用key寻找value，然后执行vlaue的过程。也就是说，一个key对应一个value，从这层意思上理解，就是if-else要干的事。那么策略模式到底优化了什么呢？其实策...

2020-07-02

509

发表评论

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。