Netty解決TCP粘包/拆包問題 - 按行分隔字符串解碼器

服務端

package org.zln.netty.five.timer;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 時間服務器服務端
 * Created by sherry on 16/11/5.
 */
public class TimerServer {
    /**
     * 服務端綁定端口號
     */
    private int PORT;

    public TimerServer(int PORT) {
        this.PORT = PORT;
    }

    /**
     * 日志
     */
    private static Logger logger = LoggerFactory.getLogger(TimerServer.class);

    public void bind() {
        /*
        NioEventLoopGroup是線程池組
        包含了一組NIO線程,專門用於網絡事件的處理
        bossGroup:服務端,接收客戶端連接
        workGroup:進行SocketChannel的網絡讀寫
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            /*
            ServerBootstrap:用於啟動NIO服務的輔助類,目的是降低服務端的開發復雜度
             */
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)//配置TCP參數,能夠設置很多,這里就只設置了backlog=1024,
                    .childHandler(new TimerServerInitializer());//綁定I/O事件處理類
            logger.debug("綁定端口號:" + PORT + ",等待同步成功");
            /*
            bind:綁定端口
            sync:同步阻塞方法,等待綁定完成,完成后返回 ChannelFuture ,主要用於通知回調
             */
            ChannelFuture channelFuture = serverBootstrap.bind(PORT).sync();
            logger.debug("等待服務端監聽窗口關閉");
            /*
             closeFuture().sync():為了阻塞,服務端鏈路關閉后才退出.也是一個同步阻塞方法
             */
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        } finally {
            logger.debug("優雅退出,釋放線程池資源");
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

package org.zln.netty.five.timer;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

/**
 * Created by sherry on 16/11/5.
 */
public class TimerServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {

        ChannelPipeline pipeline = socketChannel.pipeline();

        pipeline.addLast(new LineBasedFrameDecoder(1024));
        pipeline.addLast(new StringDecoder());
        pipeline.addLast(new TimerServerHandler());



    }
}

package org.zln.netty.five.timer;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * Handler主要用於對網絡事件進行讀寫操作,是真正的業務類
 * 通常只需要關注 channelRead 和 exceptionCaught 方法
 * Created by sherry on 16/11/5.
 */
public class TimerServerHandler extends ChannelHandlerAdapter {

    /**
     * 日志
     */
    private Logger logger = LoggerFactory.getLogger(TimerServerHandler.class);

    private static int count = 0;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        String body = (String) msg;
        logger.debug("第  "+(++count)+"  次收到請求  -  "+body);

        String timeNow = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss SSS").format(new Date())+System.lineSeparator();

        //獲取發送給客戶端的數據
        ByteBuf resBuf = Unpooled.copiedBuffer(timeNow.getBytes("UTF-8"));

        ctx.writeAndFlush(resBuf);
    }


    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //將消息發送隊列中的消息寫入到SocketChannel中發送給對方
        logger.debug("channelReadComplete");
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //發生異常時,關閉 ChannelHandlerContext,釋放ChannelHandlerContext 相關的句柄等資源
        logger.error(cause.getMessage(),cause);
        ctx.close();
    }
}

客戶端

package org.zln.netty.five.timer;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 時間服務器客戶端
 * Created by sherry on 16/11/5.
 */
public class TimerClient {
    /**
     * 日志
     */
    private Logger logger = LoggerFactory.getLogger(TimerServer.class);

    private String HOST;
    private int PORT;

    public TimerClient(String HOST, int PORT) {
        this.HOST = HOST;
        this.PORT = PORT;
    }

    public void connect(){
        //配置客戶端NIO線程組
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY,true)
                    .handler(new TimerClientInitializer());
            //發起異步連接操作
            logger.debug("發起異步連接操作 - start");
            ChannelFuture channelFuture = bootstrap.connect(HOST,PORT).sync();
            logger.debug("發起異步連接操作 - end");
            //等待客戶端鏈路關閉
            logger.debug("等待客戶端鏈路關閉 - start");
            channelFuture.channel().closeFuture().sync();
            logger.debug("等待客戶端鏈路關閉 - end");
        } catch (InterruptedException e) {
            logger.error(e.getMessage(),e);
        }finally {
            //優雅的關閉
            eventLoopGroup.shutdownGracefully();
        }
    }
}

package org.zln.netty.five.timer;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

/**
 * Created by sherry on 16/11/5.
 */
public class TimerClientInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new LineBasedFrameDecoder(1024));
        pipeline.addLast(new StringDecoder());
        pipeline.addLast(new TimerClientHandler());
    }
}

package org.zln.netty.five.timer;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;

/**
 * Created by sherry on 16/11/5.
 */
public class TimerClientHandler extends ChannelHandlerAdapter {

    /**
     * 日志
     */
    private Logger logger = LoggerFactory.getLogger(TimerClientHandler.class);

    private static int count = 0;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.debug("客戶端連接上了服務端");

        //發送請求
        ByteBuf reqBuf = null;
        for (int i = 0; i < 100; i++) {
            reqBuf = getReq("GET TIME"+System.lineSeparator());
            ctx.writeAndFlush(reqBuf);
        }


    }

    /**
     * 將字符串包裝成ByteBuf
     * @param s
     * @return
     */
    private ByteBuf getReq(String s) throws UnsupportedEncodingException {
        byte[] data = s.getBytes("UTF-8");
        ByteBuf reqBuf = Unpooled.buffer(data.length);
        reqBuf.writeBytes(data);
        return reqBuf;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String body = (String) msg;
        logger.debug("這是收到的第 "+(++count)+" 筆響應 -- "+body);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

 

這里主要使用 LineBasedFrameDecoder 和 StringDecoder 來實現解決粘包問題

原理如下：

　　LineBasedFrameDecoder 依次遍歷 ByteBuf 中的可讀字節，判斷是否有 \n 或 \r\n，如果有，就作為結束位置。從可讀索引到結束位置區間的字節組成一行。它是以換行符為結束標志的解碼器。支持攜帶結束符或者不懈怠結束符兩種解碼方式。同時支持配置單行的最大長度。如果讀取到了最大長度仍舊沒有發現換行符，就會拋出異常，同時忽略掉之前讀到的數據。

 

　　StringDecoder 的作用就是講接收到的對象轉化成字符串，然后繼續調用handler。這樣就不需要再handler中手動將對象轉化成字符串了，直接強制轉化就行。

 

　　LineBasedFrameDecoder+StringDecoder組合就是按行切割的文本解碼器，用來解決TCP的粘包和拆包問題。