码迷，mamicode.com
首页 > Web开发 > 详细

Netty中使用MessagePack时的TCP粘包问题与解决方案

时间：2018-02-15 00:18:29      阅读：364      评论：0      收藏：0      [点我收藏+]

标签：catch   pojo   ali   .so   行修改   服务   tran   sage   socket   

[toc]

Netty中使用MessagePack时的TCP粘包问题与解决方案

通过下面的实例代码来演示在Netty中使用MessagPack时会出现的TCP粘包问题，为了学习的连贯性，参考了《Netty权威指南》第7章中的代码，但是需要注意的是，书中并没有提供完整代码，提供的代码都是片段性的，所以我根据自己的理解把服务端的代码和客户端的代码写了出来，可以作为参考。

仍然需要注意的是，我使用的是Netty 4.x的版本。

另外我在程序代码中写了非常详细的注释，所以这里不再进行更多的说明。

在使用MessagePack时的TCP粘包问题

编码器与解码器

MsgpackEncoder.java
package cn.xpleaf.msgpack;

import org.msgpack.MessagePack;

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

/**
 * MsgpackEncoder继承自Netty中的MessageToByteEncoder类，
 * 并重写抽象方法encode(ChannelHandlerContext ctx, Object msg, ByteBuf out)
 * 它负责将Object类型的POJO对象编码为byte数组，然后写入到ByteBuf中
 * @author yeyonghao
 *
 */
public class MsgpackEncoder extends MessageToByteEncoder<Object> {

    @Override
    protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
        // 创建MessagePack对象
        MessagePack msgpack = new MessagePack();
        // 将对象编码为MessagePack格式的字节数组
        byte[] raw = msgpack.write(msg);
        // 将字节数组写入到ByteBuf中
        out.writeBytes(raw);
    }

}
MsgpackDecoder.java
package cn.xpleaf.msgpack;

import java.util.List;

import org.msgpack.MessagePack;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToMessageDecoder;

/**
 * MsgpackDecoder继承自Netty中的MessageToMessageDecoder类，
 * 并重写抽象方法decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out)
 * 首先从数据报msg（数据类型取决于继承MessageToMessageDecoder时填写的泛型类型）中获取需要解码的byte数组
 * 然后调用MessagePack的read方法将其反序列化（解码）为Object对象
 * 将解码后的对象加入到解码列表out中，这样就完成了MessagePack的解码操作
 * @author yeyonghao
 *
 */
public class MsgpackDecoder extends MessageToMessageDecoder<ByteBuf> {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
        // 从数据报msg中（这里的数据类型为ByteBuf，因为Netty的通信基于ByteBuf对象）
        final byte[] array;
        final int length = msg.readableBytes();
        array = new byte[length];
        /**
         * 这里使用的是ByteBuf的getBytes方法来将ByteBuf对象转换为字节数组，前面是使用readBytes，直接传入一个接收的字节数组参数即可
         * 这里的参数比较多，第一个参数是index，关于readerIndex，说明如下：
         * ByteBuf是通过readerIndex跟writerIndex两个位置指针来协助缓冲区的读写操作的，具体原理等到Netty源码分析时再详细学习一下
         * 第二个参数是接收的字节数组
         * 第三个参数是dstIndex the first index of the destination
         * 第四个参数是length   the number of bytes to transfer
         */
        msg.getBytes(msg.readerIndex(), array, 0, length);
        // 创建一个MessagePack对象
        MessagePack msgpack = new MessagePack();
        // 解码并添加到解码列表out中
        out.add(msgpack.read(array));
    }

}

服务端

EchoServer.java
package cn.xpleaf.echo;

import cn.demo.simple.MsgPackDecode;
import cn.xpleaf.msgpack.MsgpackDecoder;
import cn.xpleaf.msgpack.MsgpackEncoder;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class EchoServer {
    public void bind(int port) throws Exception {
        // 配置服务端的NIO线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // 添加MesspagePack解码器
                        ch.pipeline().addLast("msgpack decoder", new MsgPackDecode());
                        // 添加MessagePack编码器
                        ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder());
                        // 添加业务处理handler
                        ch.pipeline().addLast(new EchoServerHandler());
                    }
                });

            // 绑定端口，同步等待成功
            ChannelFuture f = b.bind(port).sync();

            // 等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 8080;
        if(args != null && args.length > 0) {
            try {
                port = Integer.valueOf(port);
            } catch (NumberFormatException e) {
                // TODO: handle exception
            }
        }
        new EchoServer().bind(port);
    }
}
EchoServerHandler.java
package cn.xpleaf.echo;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("Server receive the msgpack message : " + msg);
        ctx.write(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 发生异常，关闭链路
        ctx.close();
    }
}

客户端

EchoClient.java
package cn.xpleaf.echo;

import cn.demo.simple.MsgPackDecode;
import cn.xpleaf.msgpack.MsgpackDecoder;
import cn.xpleaf.msgpack.MsgpackEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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 EchoClient {
    public void connect(String host, int port, int sendNumber) throws Exception {
        // 配置客户端NIO线程组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                // 设置TCP连接超时时间
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                .handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // 添加MesspagePack解码器
                        ch.pipeline().addLast("msgpack decoder", new MsgPackDecode());
                        // 添加MessagePack编码器
                        ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder());
                        // 添加业务处理handler
                        ch.pipeline().addLast(new EchoClientHandler(sendNumber));
                    }
                });
            // 发起异步连接操作
            ChannelFuture f = b.connect(host, port).sync();

            // 等待客户端链路关闭
            f.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放NIO线程组
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 8080;
        if(args != null && args.length > 0) {
            try {
                port = Integer.valueOf(port);
            } catch (NumberFormatException e) {
                // 采用默认值
            }
        }
        int sendNumber = 1000;
        new EchoClient().connect("localhost", port, sendNumber);
    }
}
EchoClientHander.java
package cn.xpleaf.echo;

import cn.xpleaf.pojo.User;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class EchoClientHandler extends ChannelInboundHandlerAdapter {

    // sendNumber为写入发送缓冲区的对象数量
    private int sendNumber;

    public EchoClientHandler(int sendNumber) {
        this.sendNumber = sendNumber;
    }

    /**
     * 构建长度为userNum的User对象数组
     * @param userNum
     * @return
     */
    private User[] getUserArray(int userNum) {
        User[] users = new User[userNum];
        User user = null;
        for(int i = 0; i < userNum; i++) {
            user = new User();
            user.setName("ABCDEFG --->" + i);
            user.setAge(i);
            users[i] = user;
        }
        return users;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        User[] users = getUserArray(sendNumber);
        for (User user : users) {
            ctx.writeAndFlush(user);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("Client receive the msgpack message : " + msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

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

}

POJO

User.java
package cn.xpleaf.pojo;

import org.msgpack.annotation.Message;

@Message
public class User {
    private String name;
    private int age;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    @Override
    public String toString() {
        return "User [name=" + name + ", age=" + age + "]";
    }

}

测试

当EchoClient.java中的sendNumber为1时，服务端和客户端都是正常工作的，此时，服务端和客户端的输出分别如下：

服务端：

Server receive the msgpack message : ["ABCDEFG --->0",0]

客户端：

Client receive the msgpack message : ["ABCDEFG --->0",0]

但是当sendNumber数字很大时，就不能正常工作了，比如可以设置为1000，此时输出结果如下：

服务端：

Server receive the msgpack message : ["ABCDEFG --->0",0]
Server receive the msgpack message : ["ABCDEFG --->1",1]
Server receive the msgpack message : ["ABCDEFG --->3",3]
...省略输出...
Server receive the msgpack message : ["ABCDEFG --->146",146]
Server receive the msgpack message : 70
Server receive the msgpack message : ["ABCDEFG --->156",156]
Server receive the msgpack message : ["ABCDEFG --->157",157]
...省略输出...

客户端：

Client receive the msgpack message : ["ABCDEFG --->0",0]
Client receive the msgpack message : 62
Client receive the msgpack message : 68

显然运行结果跟预期的不太一样，这是因为出现了TCP粘包问题。

粘包问题解决方案

在前面代码的基础上，只需要对EchoServer.java和EchoClient.java中的代码进行修改即可。

EchoServer.java

package cn.xpleaf.echo02;

import cn.demo.simple.MsgPackDecode;
import cn.xpleaf.msgpack.MsgpackDecoder;
import cn.xpleaf.msgpack.MsgpackEncoder;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;

public class EchoServer {
    public void bind(int port) throws Exception {
        // 配置服务端的NIO线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // 添加长度字段解码器
                        // 在MessagePack解码器之前增加LengthFieldBasedFrameDecoder，用于处理半包消息
                        // 它会解析消息头部的长度字段信息，这样后面的MsgpackDecoder接收到的永远是整包消息
                        ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2));
                        // 添加MesspagePack解码器
                        ch.pipeline().addLast("msgpack decoder", new MsgPackDecode());
                        // 添加长度字段编码器
                        // 在MessagePack编码器之前增加LengthFieldPrepender，它将在ByteBuf之前增加2个字节的消息长度字段
                        ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(2));
                        // 添加MessagePack编码器
                        ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder());
                        // 添加业务处理handler
                        ch.pipeline().addLast(new EchoServerHandler());
                    }
                });

            // 绑定端口，同步等待成功
            ChannelFuture f = b.bind(port).sync();

            // 等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 8080;
        if(args != null && args.length > 0) {
            try {
                port = Integer.valueOf(port);
            } catch (NumberFormatException e) {
                // TODO: handle exception
            }
        }
        new EchoServer().bind(port);
    }
}

EchoClient.java

package cn.xpleaf.echo02;

import cn.demo.simple.MsgPackDecode;
import cn.xpleaf.msgpack.MsgpackDecoder;
import cn.xpleaf.msgpack.MsgpackEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;

public class EchoClient {
    public void connect(String host, int port, int sendNumber) throws Exception {
        // 配置客户端NIO线程组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                // 设置TCP连接超时时间
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                .handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // 添加长度字段解码器
                        // 在MessagePack解码器之前增加LengthFieldBasedFrameDecoder，用于处理半包消息
                        // 它会解析消息头部的长度字段信息，这样后面的MsgpackDecoder接收到的永远是整包消息
                        ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2));
                        // 添加MesspagePack解码器
                        ch.pipeline().addLast("msgpack decoder", new MsgPackDecode());
                        // 添加长度字段编码器
                        // 在MessagePack编码器之前增加LengthFieldPrepender，它将在ByteBuf之前增加2个字节的消息长度字段
                        ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(2));
                        // 添加MessagePack编码器
                        ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder());
                        // 添加业务处理handler
                        ch.pipeline().addLast(new EchoClientHandler(sendNumber));
                    }
                });
            // 发起异步连接操作
            ChannelFuture f = b.connect(host, port).sync();

            // 等待客户端链路关闭
            f.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放NIO线程组
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 8080;
        if(args != null && args.length > 0) {
            try {
                port = Integer.valueOf(port);
            } catch (NumberFormatException e) {
                // 采用默认值
            }
        }
        int sendNumber = 1000;
        new EchoClient().connect("localhost", port, sendNumber);
    }
}

测试

可以将EchoClient.java中sendNumber设置为1000或更大，此时服务端和客户端的输出结果跟预期的都是一样的。

测试结果为，服务端和客户端都会打印1000行的信息（假设sendNumber为1000），这里不再给出运行结果。

Netty中使用MessagePack时的TCP粘包问题与解决方案

标签：catch   pojo   ali   .so   行修改   服务   tran   sage   socket   

原文地址：http://blog.51cto.com/xpleaf/2071688
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！