Netty中序列化架構MessagePack的簡單實作

 MessagePack是一個高效的二進制序列化架構，它像JSON一樣支援不同語言間的資料交換，但是它的性能更快，序列化之後的碼流也更小。MessagePack的特點如下:

   編解碼高效，性能高；

   序列化之後碼流小

   支援跨語言

MessagePack使用

1.依賴

 使用maven建構項目

<dependency>
    <groupId>org.msgpack</groupId>
    <artifactId>msgpack</artifactId>
    <version>0.6.12</version>
</dependency>      

2.建立編碼和解碼器

編碼器

/**
 * @param ctx 上下文
 * @param msg 需要編碼的對象
 * @param out 編碼後的資料
 */
@Override
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
    
    MessagePack msgpack = new MessagePack();
    // 對對象進行序列化
    byte[] raw = msgpack.write(msg);
    // 傳回序列化的資料
    out.writeBytes(raw);
}      

解碼器

/**
 * @param ctx 上下文
 * @param msg 需要解碼的資料
 * @param out 解碼清單
 */
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
    final byte[] array;
    final int length = msg.readableBytes();
    array = new byte[length];
    // 擷取需要解碼的位元組數組
    msg.getBytes(msg.readerIndex(), array,0,length);
    MessagePack msgpack = new MessagePack();
    // 反序列化并将結果儲存到了解碼清單中
    out.add(msgpack.read(array));
}      

3.用戶端

EchoClient

/**
 * MsgPack 編解碼器
 * 
 * @author 波波烤鴨
 * @email [email protected]
 *
 */
public class EchoClient {

    public static void main(String[] args) throws Exception {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                // 采用預設值
            }
        }
        new EchoClient().connector(port, "127.0.0.1",10);
    }

    public void connector(int port, String host,final 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)
                            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                            .handler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    //這裡設定通過增加標頭表示封包長度來避免粘包
                    ch.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(1024, 0, 2,0,2));
                    //增加解碼器
                    ch.pipeline().addLast("msgpack decoder",new MsgpackDecoder());
                    //這裡設定讀取封包的標頭長度來避免粘包
                    ch.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));
                    //增加編碼器
                    ch.pipeline().addLast("msgpack encoder",new MsgpackEncoder());
                    // 4.添加自定義的處理器
                    ch.pipeline().addLast(new EchoClientHandler(sendNumber));
                }
            });

            // 發起異步連接配接操作
            ChannelFuture f = b.connect(host, port).sync();
            // 等待用戶端鍊路關閉
            f.channel().closeFuture().sync();
        }catch(Exception e){
            e.printStackTrace();
        }  finally {
            // 優雅退出，釋放NIO線程組
            group.shutdownGracefully();
        }
    }

}      

EchoClientHandler

/**
 * DelimiterBasedFrameDecoder 案例
 *      自定義處理器
 * @author 波波烤鴨
 * @email [email protected]
 *
 */
public class EchoServerHandler extends ChannelHandlerAdapter{


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //UserInfo user = (UserInfo) msg;
        System.out.println("server receive the msgpack message :"+msg);
        //ctx.writeAndFlush(user);
        ctx.writeAndFlush(msg);
        
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close(); // 發生異常關閉鍊路
    }
}      

4.服務端

EchoServer

/**
 * MsgPack 編解碼器
 *      服務端
 * @author 波波烤鴨
 * @email [email protected]
 *
 */
public class EchoServer {

    public void bind(int port) throws Exception {
        // 配置服務端的NIO線程組
        // 服務端接受用戶端的連接配接
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        // 進行SocketChannel的網絡讀寫
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 100)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(65535, 0, 2,0,2));
                            // 添加msgpack的編碼和解碼器
                            ch.pipeline().addLast("msgpack decoder",new MsgpackDecoder());
                            ch.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));
                            ch.pipeline().addLast("msgpack encoder",new MsgpackEncoder());
                            // 添加自定義的處理器
                            ch.pipeline().addLast(new EchoServerHandler());

                        }
                    });

            // 綁定端口，同步等待成功
            ChannelFuture f = b.bind(port).sync();
            // 等待服務端監聽端口關閉
            f.channel().closeFuture().sync();
        }catch(Exception e){
            e.printStackTrace();
        } 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(args[0]);
            }catch(NumberFormatException e){
                // 采用預設值
            }
        }
        new EchoServer().bind(port);
    }

}      

EchoServerHandler

/**
 * DelimiterBasedFrameDecoder 案例
 *      自定義處理器
 * @author 波波烤鴨
 * @email [email protected]
 *
 */
public class EchoServerHandler extends ChannelHandlerAdapter{


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //UserInfo user = (UserInfo) msg;
        System.out.println("server receive the msgpack message :"+msg);
        //ctx.writeAndFlush(user);
        ctx.writeAndFlush(msg);
        
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close(); // 發生異常關閉鍊路
    }
}      

5.注意點（POJO）

 消息類上加上注解Message，還有就是必須要有預設的無參構造器

/**
 * Msgpack 中必須添加@Message注解 及 無參構造方法
 * @author 波波烤鴨
 * @email [email protected]
 *
 */
@Message
public class UserInfo {


    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 "UserInfo [name=" + name + ", age=" + age + "]";
    }
}      

6.測試

服務端輸出

server receive the msgpack message :["bobo烤鴨:0",0]
server receive the msgpack message :["bobo烤鴨:1",1]
server receive the msgpack message :["bobo烤鴨:2",2]
server receive the msgpack message :["bobo烤鴨:3",3]
server receive the msgpack message :["bobo烤鴨:4",4]
server receive the msgpack message :["bobo烤鴨:5",5]
server receive the msgpack message :["bobo烤鴨:6",6]
server receive the msgpack message :["bobo烤鴨:7",7]
server receive the msgpack message :["bobo烤鴨:8",8]
server receive the msgpack message :["bobo烤鴨:9",9]      

用戶端輸出

Client receive the msgpack message :["bobo烤鴨:0",0]
Client receive the msgpack message :["bobo烤鴨:1",1]
Client receive the msgpack message :["bobo烤鴨:2",2]
Client receive the msgpack message :["bobo烤鴨:3",3]
Client receive the msgpack message :["bobo烤鴨:4",4]
Client receive the msgpack message :["bobo烤鴨:5",5]
Client receive the msgpack message :["bobo烤鴨:6",6]
Client receive the msgpack message :["bobo烤鴨:7",7]
Client receive the msgpack message :["bobo烤鴨:8",8]
Client receive the msgpack message :["bobo烤鴨:9",9]      

至此Netty中就可以通過MessagePack來處理序列化的情況了~