概述

在

三篇文章中,筆者介紹了在Netty如何解決拆包和粘包問題,其中自定義解碼器處理半包訊息 裡面介紹的方法是在線上實際用過的，經過了超級大流量的驗證，挺靠譜的。下面再介紹另外一種解決半包問題的方法，雖然我沒實際在線上用過，但是可以當成是一種知識補充。

直接上程式碼

在上面提到的三篇文章中，已經詳細介紹了拆包和粘包的問題，這裡不在詳細說了。下面我就直接把ReplayingDecoder的程式碼貼一下,這些程式碼都是可以直接執行的。

DecoderServer

package netty;
import io.netty.bootstrap.ServerBootstrap
 
;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class DecoderServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1
 
);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).
                    childHandler(new DecoderServerInitializer());
 


            ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

DecoderServerInitializer

在這裡需要加入我們自定義的編碼器和解碼器。

package netty;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;


public class DecoderServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        pipeline.addLast(new UserDecoder());
        pipeline.addLast(new UserEncoder());

        pipeline.addLast(new DecoderServerHandler());
    }
}

UserDecoder

繼承自ReplayingDecoder。

package netty;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;

import java.util.List;

public class UserDecoder extends ReplayingDecoder<Void> {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        int length = in.readInt();

        byte[] content = new byte[length];
        in.readBytes(content);

        UserProtocol personProtocol = new UserProtocol();
        personProtocol.setLength(length);
        personProtocol.setContent(content);

        out.add(personProtocol);
    }
}

UserEncoder

package netty;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

public class UserEncoder extends MessageToByteEncoder<UserProtocol> {

    @Override
    protected void encode(ChannelHandlerContext ctx, UserProtocol msg, ByteBuf out) throws Exception {
        out.writeInt(msg.getLength());
        out.writeBytes(msg.getContent());
    }
}

UserProtocol

自定義協議。

package netty;

public class UserProtocol {

    private int length;

    private byte[] content;

    public int getLength() {
        return length;
    }

    public void setLength(int length) {
        this.length = length;
    }

    public byte[] getContent() {
        return content;
    }

    public void setContent(byte[] content) {
        this.content = content;
    }
}

DecoderServerHandler

package netty;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.Charset;
import java.util.UUID;

public class DecoderServerHandler extends SimpleChannelInboundHandler<UserProtocol> {

    private int count;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, UserProtocol msg) throws Exception {
        int length = msg.getLength();
        byte[] content = msg.getContent();

        System.out.println("服務端接收到的訊息：" + new String(content, Charset.forName("utf-8")));

        System.out.println("服務端接收到的訊息數量：" + (++this.count));

        String responseMessage = UUID.randomUUID().toString();
        int responseLength = responseMessage.getBytes("utf-8").length;
        byte[] responseContent = responseMessage.getBytes("utf-8");

        UserProtocol userProtocol = new UserProtocol();
        userProtocol.setLength(responseLength);
        userProtocol.setContent(responseContent);

        ctx.writeAndFlush(userProtocol);
    }

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

下面貼一下Netty客戶端的程式碼。

DecoderClient

package netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class DecoderClient {

    public static void main(String[] args) throws Exception {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class).
                    handler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();

                            pipeline.addLast(new UserDecoder());
                            pipeline.addLast(new UserEncoder());

                            pipeline.addLast(new DecoderClientHandler());
                        }
                    });

            ChannelFuture channelFuture = bootstrap.connect("localhost", 8899).sync();
            channelFuture.channel().closeFuture().sync();
        } finally {
            eventLoopGroup.shutdownGracefully();
        }
    }
}

DecoderClientHandler

package netty;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.Charset;

public class DecoderClientHandler extends SimpleChannelInboundHandler<UserProtocol> {

    private int count;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 3; ++i) {
            String messageToBeSent = "sent from client ";
            byte[] content = messageToBeSent.getBytes(Charset.forName("utf-8"));
            int length = messageToBeSent.getBytes(Charset.forName("utf-8")).length;

            UserProtocol userProtocol = new UserProtocol();
            userProtocol.setLength(length);
            userProtocol.setContent(content);

            ctx.writeAndFlush(userProtocol);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, UserProtocol msg) throws Exception {
        int length = msg.getLength();
        byte[] content = msg.getContent();

        System.out.println("客戶端接收到的內容：" + new String(content, Charset.forName("utf-8")));

        System.out.println("客戶端接受到的訊息數量：" + (++this.count));
    }

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

執行結果

服務端

服務端接收到的訊息：sent from client 
服務端接收到的訊息數量：1
服務端接收到的訊息：sent from client 
服務端接收到的訊息數量：2
服務端接收到的訊息：sent from client 
服務端接收到的訊息數量：3

客戶端

客戶端接收到的內容：3879cc39-2bfd-4ca4-826a-91658bfcd97d
客戶端接受到的訊息數量：1
客戶端接收到的內容：a63108e6-4e45-41e2-b78b-31228461e3ce
客戶端接受到的訊息數量：2
客戶端接收到的內容：f760b6c1-5c42-4fa7-8fa2-32b1941c7f9d
客戶端接受到的訊息數量：3

可見用ReplayingDecoder 可以解決拆包和粘包的問題。