『互联网架构』软件架构-netty粘包分包编码解码（57）

一般直接接触RPC框架的时候内部都做了对于粘包分包的解决方案，咱们来一起了解下这方便的含义，包括编码解码这块。

源码：https://github.com/limingios/netFuture/tree/master/源码/『互联网架构』软件架构-io与nio线程模型reactor模型（上）（53）/nio

（一）粘包分包概念

• 粘包

  > TCP

  由于TCP协议本身的机制（面向连接的可靠地协议-三次握手机制）客户端与服务器会维持一个连接（Channel），数据在连接不断开的情况下，可以持续不断地将多个数据包发往服务器，但是如果发送的网络数据包太小，那么他本身会启用Nagle算法（可配置是否启用）对较小的数据包进行合并（基于此，TCP的网络延迟要UDP的高些）然后再发送（超时或者包大小足够）。那么这样的话，服务器在接收到消息（数据流）的时候就无法区分哪些数据包是客户端自己分开发送的，这样产生了粘包；服务器在接收到数据库后，放到缓冲区中，如果消息没有被及时从缓存区取走，下次在取数据的时候可能就会出现一次取出多个数据包的情况，造成粘包现象（确切来讲，对于基于TCP协议的应用，不应用包来描述，而应 用 流来描述），个人认为服务器接收端产生的粘包应该与 linux 内核处理socket的方式 select轮询机制的线性扫描频度无关。

UDP

本身作为无连接的不可靠的传输协议（适合频繁发送较小的数据包），他不会对数据包进行合并发送（也就没有Nagle算法之说了），他直接是一端发送什么数据，直接就发出去了，既然他不会对数据合并，每一个数据包都是完整的（数据+UDP头+IP头等等发一次数据封装一次）也就没有粘包一说了。

• 分包

  >可能是IP分片传输导致的，也可能是传输过程中丢失部分包导致出现的半包，还有可能就是一个包可能被分成了两次传输，在取数据的时候，先取到了一部分（还可能与接收的缓冲区大小有关系），总之就是一个数据包被分成了多次接收。

• TCP当中,只有流的概念,没有包的概念（根本原因）

  >简单的概括

（1）粘包：

1.服务端

原因收到的数据放在系统接收缓冲区，用户进程从该缓冲区取数据

2.客户端

原因TCP为提高传输效率，要收集到足够多的数据后才发送一包数据

（2）.分包：

1.应用程序写入的字节大小大于套接字发送缓冲区的大小

2.进行mss（最大报文长度）大小的TCP分段，当TCP报文长度-TCP头部长度>MSS

3.以太网帧的payload（净荷）大于MTU（1500字节）进行ip分片

（二）Netty粘包分包现象演示

源码：pack目录下的error

Server.java

package com.dig8.pack.error;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * netty服务端
 * 
 * @author idig8.com
 */
public class Server {

    public static void main(String[] args) {
        //  服务类
        ServerBootstrap bootstrap = new ServerBootstrap();
        //  boss线程，主要监听端口和获取worker线程及分配socketChannel给worker线程
        ExecutorService boss = Executors.newCachedThreadPool();
        //  worker线程负责数据读写
        ExecutorService worker = Executors.newCachedThreadPool();
        //  设置niosocket工厂
        bootstrap.setFactory(new NioServerSocketChannelFactory(boss, worker));
        //  设置管道的工厂
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() throws Exception {
                ChannelPipeline pipeline = Channels.pipeline();
                // 管道过滤器
                pipeline.addLast("myHandler", new ServerHandler());
                return pipeline;
            }
        });

        // 服务类绑定端口
        bootstrap.bind(new InetSocketAddress(8888));
    }
}

 ServerHandler.java 

package com.dig8.pack.error;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

/**
 * @author idig8.com
 */
public class ServerHandler extends SimpleChannelHandler{

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
        byte[] bs = buffer.array();
        System.out.println("server receive data: " +new String(bs));
    }
}

 Client.java 

package com.dig8.pack.error;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.string.StringEncoder;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 客户端
 * 
 * @author idig8.com
 */
public class Client {

    public static void main(String[] args) throws Exception {

        //服务类
        ClientBootstrap bootstrap = new  ClientBootstrap();
        //线程池
        ExecutorService boss = Executors.newCachedThreadPool();
        ExecutorService worker = Executors.newCachedThreadPool();
        //socket工厂
        bootstrap.setFactory(new NioClientSocketChannelFactory(boss, worker));
        //管道工厂
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() throws Exception {
                ChannelPipeline pipeline = Channels.pipeline();
                pipeline.addLast("1", new StringEncoder());
                pipeline.addLast("2", new ClientHandler());
                return pipeline;
            }
        });

        //连接服务端
        bootstrap.connect(new InetSocketAddress("127.0.0.1", 8888)).sync();
    }

}

 ClientHandler.java 

package com.dig8.pack.error;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.*;

/**
 * 客户端消息处理类
 * @author idig8.com
 */
public class ClientHandler extends SimpleChannelHandler {
    // 包头
    private static final int HEAD_FLAG = -32323231;
    @Override
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        Channel channel = ctx.getChannel();
        String msg = "Hello,idig8.com";

        for (int i = 0; i < 1000; i++) {
            channel.write(msg);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
    }
}

运行后出现粘包和分包现象

（三）粘包分包问题解决思路

服务端和客户端约定好稳定的数据包结构

1.客户端根据约定的数据包结构发送数据

2.服务端根据约定的数据包结构来读取数据

通过MyDecoder集成FrameDecoder的方式来

源码：pack目录下的custom

Server.java

package com.dig8.pack.custom;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
/**
 * netty服务端
 * 
 * @author idig8.com
 */
public class Server {

    public static void main(String[] args) {
        //  服务类
        ServerBootstrap bootstrap = new ServerBootstrap();
        //  boss线程，主要监听端口和获取worker线程及分配socketChannel给worker线程
        ExecutorService boss = Executors.newCachedThreadPool();
        //  worker线程负责数据读写
        ExecutorService worker = Executors.newCachedThreadPool();
        //  设置niosocket工厂
        bootstrap.setFactory(new NioServerSocketChannelFactory(boss, worker));
        //  设置管道的工厂
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() throws Exception {
                ChannelPipeline pipeline = Channels.pipeline();
                // 管道过滤器
                pipeline.addLast("myDecoder", new MyDecoder());
                pipeline.addLast("myHandler", new ServerHandler());
                return pipeline;
            }
        });

        // 服务类绑定端口
        bootstrap.bind(new InetSocketAddress(7778));
    }
}

 ServerHandler.java 

package com.dig8.pack.custom;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

/**
 * @author idig8.com
 */
public class ServerHandler extends SimpleChannelHandler{

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        /*ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
        byte[] bs = buffer.array();*/
        System.out.println("server receive data: " + e.getMessage());
    }
}

 Client.java 

package com.dig8.pack.custom;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.string.StringEncoder;
/**
 * 客户端
 * 
 * @author idig8.com
 */
public class Client {

    public static void main(String[] args) throws Exception {

        //服务类
        ClientBootstrap bootstrap = new  ClientBootstrap();
        //线程池
        ExecutorService boss = Executors.newCachedThreadPool();
        ExecutorService worker = Executors.newCachedThreadPool();
        //socket工厂
        bootstrap.setFactory(new NioClientSocketChannelFactory(boss, worker));
        //管道工厂
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() throws Exception {
                ChannelPipeline pipeline = Channels.pipeline();
                pipeline.addLast("1", new StringEncoder());
                pipeline.addLast("2", new ClientHandler());
                return pipeline;
            }
        });

        //连接服务端
        bootstrap.connect(new InetSocketAddress("127.0.0.1", 7778)).sync();
    }

}

 ClientHandler.java 

package com.dig8.pack.custom;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

/**
 * 客户端消息处理类
 * @author idig8.com
 */
public class ClientHandler extends SimpleChannelHandler {
    // 包头
    private static final int HEAD_FLAG = -32323231;
    @Override
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        Channel channel = ctx.getChannel();
        String msg = "Hello,idig8.com 通过定义包头+长度+数据 防止粘包和分包";
        byte[] bytes = msg.getBytes();
        // 定义数据包 ，结构为：包头 + 长度 + 数据
        ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();
        // 1.写包头
        buffer.writeInt(HEAD_FLAG);// 4字节
        // 2.写长度
        buffer.writeInt(bytes.length);// 4字节
        // 3.写数据本身
        buffer.writeBytes(bytes);

        for (int i = 0; i < 1000; i++) {
            channel.write(buffer);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
    }
}

 MyDecoder.java 

/**
 * 
 */
package com.dig8.pack.custom;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.frame.FrameDecoder;

/**
 * @author idig8.com
 */
public class MyDecoder extends FrameDecoder{
    // 包头
    private static final int HEAD_FLAG = -32323231;
    // 数据包基本长度
    private final static int BASE_LENGTH = 4 + 4;
    @Override
    protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception {

        // 收到数据之后，先判断buffer中可读的数据长度是否大于数据包的基本长度
        if(buffer.readableBytes() > BASE_LENGTH){
            // 防止socket攻击：
            if(buffer.readableBytes() > 4096 * 2){ // 4k
                System.out.println("socket 攻击了");
                buffer.skipBytes(buffer.readableBytes());
            }
            // 记录包头开始的位置
            int headIndex;
            while(true){
                headIndex = buffer.readerIndex();
                buffer.markReaderIndex();
                // 代码很关键
                if(buffer.readableBytes() < 4){// 包头的长度
                    buffer.readerIndex(headIndex);
                    return null;
                }

                // 此时说明包头的长度是足够的
                // 正好读取的是包头
                if(buffer.readInt() == HEAD_FLAG ){
                    break;
                }

                // [1,2,3,4] 1 1 1 1 1
                // 如果不是包头,需要略过一个字节，在略过之前，需要还原读指针位置
                buffer.resetReaderIndex();
                buffer.readByte();// 略过一个字节

                if(buffer.readableBytes() < BASE_LENGTH){
                    return null;
                }

            }

            // 此时说明有数据包到来
            // 做标记(记住当前读指针的位置)
            // buffer.markReaderIndex();

            // 1.读长度
            int dataLength = buffer.readInt();

            if(buffer.readableBytes() < dataLength){
                // 说明数据本身的长度还不够， 肯定要继续等待后面的数据到来
                // 还原读指针的位置
                buffer.readerIndex(headIndex);
                return null;
            }

            // 此时说明数据包已经位置
            // 2.读数据本身
            byte[] dst = new byte[dataLength];
            buffer.readBytes(dst);
            // 继续传递下去
            // ?
            // 如果此时buffer中的数据还没有读完，那么剩下的数据怎么办？
            return new String(dst);

        }
        // return null 表示此时的数据包不完整，需要继续等待下一个数据包的到来 ？
        return null;
    }

}

（三）Netty自带粘包分包解决方案

消息定长

1.FixedLengthFrameDecoder

行分隔符

2.LineBasedFrameDecoder

自定义特殊符号进行分割

3.DelimiterBasedFrameDecoder

源码：pack目录下的nettysolution

Server.java

package com.dig8.pack.nettysolution;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.frame.DelimiterBasedFrameDecoder;
import org.jboss.netty.handler.codec.frame.FixedLengthFrameDecoder;
import org.jboss.netty.handler.codec.frame.LineBasedFrameDecoder;
import org.jboss.netty.handler.codec.string.StringDecoder;
/**
 * 服务端
 * @author idig8.com
 */
public class Server {
    public static void main(String[] args) throws Exception {
        // 服务类
        ServerBootstrap bootstrap = new ServerBootstrap();

        // boss线程，主要监听端口和获取worker线程及分配socketChannel给worker线程
        ExecutorService boss = Executors.newCachedThreadPool();
        // worker线程负责数据读写
        ExecutorService worker = Executors.newCachedThreadPool();

        // 设置niosocket工厂
        bootstrap.setFactory(new NioServerSocketChannelFactory(boss, worker));

        // 设置管道的工厂
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() throws Exception {
                ChannelPipeline pipeline = Channels.pipeline();
                // 管道过滤器
                // 方案1：消息定长
                //pipeline.addLast("fixedLength", new FixedLengthFrameDecoder(18));

                // 方案2：行分隔符
                //pipeline.addLast("fixedLength", new LineBasedFrameDecoder(1024));

                // 方案3：自定义特殊符号进行分割
                pipeline.addLast("delimiter", new DelimiterBasedFrameDecoder(1024, 
                        ChannelBuffers.copiedBuffer("#@#".getBytes())));
                pipeline.addLast("1",new StringDecoder());  
                pipeline.addLast("2",new ServerMessageHandler());       
                return pipeline;
            }
        });

        // 服务类绑定端口
        bootstrap.bind(new InetSocketAddress(7777));

        System.out.println("服务端启动...");

    }


}

 ServerMessageHandler.java 

package com.dig8.pack.nettysolution;

import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
/**
 * 服务端消息处理类
 * @author idig8.com
 */
public class ServerMessageHandler extends SimpleChannelHandler {

    /**
     * 接收消息
     */
    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        System.out.println("receive request: " + e.getMessage());
    }

    /**
     * 异常处理
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
    }
}

 Client.java 

package com.dig8.pack.nettysolution;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.frame.DelimiterBasedFrameDecoder;
import org.jboss.netty.handler.codec.frame.FixedLengthFrameDecoder;
import org.jboss.netty.handler.codec.frame.LineBasedFrameDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;
/**
 * 客户端
 * 
 * @author idig8.com
 */
public class Client {

    public static void main(String[] args) throws Exception {

        //服务类
        ClientBootstrap bootstrap = new  ClientBootstrap();

        //线程池
        ExecutorService boss = Executors.newCachedThreadPool();
        ExecutorService worker = Executors.newCachedThreadPool();

        //socket工厂
        bootstrap.setFactory(new NioClientSocketChannelFactory(boss, worker));

        //管道工厂
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {

            @Override
            public ChannelPipeline getPipeline() throws Exception {
                ChannelPipeline pipeline = Channels.pipeline();
                // 方案1：消息定长
                //pipeline.addLast("fixedLength", new FixedLengthFrameDecoder(18));
                // 方案2：行分隔符
                //pipeline.addLast("fixedLength", new LineBasedFrameDecoder(1024));
                // 方案3：自定义特殊符号进行分割
                pipeline.addLast("delimiter", new DelimiterBasedFrameDecoder(1024, 
                        ChannelBuffers.copiedBuffer("#@#".getBytes())));
                pipeline.addLast("1",new StringEncoder());
                pipeline.addLast("2", new ClientMessageHandler());
                return pipeline;
            }
        });

        //连接服务端
        @SuppressWarnings("unused")
        ChannelFuture connect = bootstrap.connect(new InetSocketAddress("127.0.0.1", 7777)).sync();


//      Channel channel = connect.getChannel();
//      System.out.println("client start");

//      Scanner scanner = new Scanner(System.in);
//      while(true){
//          System.out.println("请输入:");
//          channel.write(scanner.next());
//      }
    }

}

 ClientMessageHandler.java 

package com.dig8.pack.nettysolution;

import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

/**
 * 客户端消息接受处理类
 * @author  idig8.com
 */
public class ClientMessageHandler extends SimpleChannelHandler {

    /**
     * 接收消息
     */
    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        System.out.println("server response : " + e.getMessage());
    }

    /**
     * 新连接
     */
    @Override
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        Channel channel = ctx.getChannel();
        String separator = "#@#";//System.getProperty("line.separator");// 系统换行符
        String msg = "idig8.com send cmd";
        for (int i = 0; i < 1000; i++) {
            channel.write(msg + i + separator);
        }
    }

    /**
     * 异常处理
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
    }
}

PS：基本上netty针对tcp 分包粘包已经说完了，确实有了netty真的很方便比传统的socket方便很多。下次说说http 协议实现。

百度未收录

>>原创文章，欢迎转载。转载请注明：转载自IT人故事会，谢谢！

已是最新文章