『互联网架构』软件架构-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人故事会,谢谢!
>>原文链接地址:『互联网架构』软件架构-netty粘包分包编码解码(57)
上一篇: 下一篇:

发表评论

电子邮件地址不会被公开。 必填项已用*标注