首页 > Web开发 > 详细

MessagePack在Netty中的应用

时间:2018-02-14 23:56:57      阅读:202      评论:0      收藏:0      [点我收藏+]

标签:泛型类   time   缓冲   cli   程序   cep   his   t对象   学习   

[toc]


MessagePack在Netty中的应用

前面使用Netty通信时,传输的都是字符串对象,因为在进行远程过程调用时,更多的是传输pojo对象,这时就需要对pojo对象进行序列化与反序列化(编码与解码),因为Java序列化技术本身的局限性,所以往往会使用第三方的编解码框架,如这里使用的MessagePack。

在使用MessagePack时,需要注意下面两点:

  • MessagePack编码后的结果是一个List对象;
  • 传输的pojo对象一定要加上@Message注解,否则无法使用MessagePack进行编码;

上面两点确实非常重要,我第一次在Netty中使用MessagePack,因为没有注意上面两点,写的Netty程序运行没有报错,客户端连接服务端也没有问题,但就是不能输出传输的pojo对象,原因就是上面的这两个问题,所以一定要先知道这两点原理,否则后面在测试Netty程序时会有很多问题,并且排错debug过程也不太容易。

下面就直接给出demo的代码,因为在代码中都加了很多注释,所以这里不再详细进行说明。

编码器与解码器

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

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));
    }

}

服务端

PojoServer.java

package cn.xpleaf.basic;

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 PojoServer {

    public void bind(int port) throws Exception {
        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 MsgpackDecoder());
                        // 添加MessagePack编码器
                        ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder());
                        // 添加业务处理handler
                        ch.pipeline().addLast(new PojoServerHandler());
                    }
                });

            // 绑定端口,同步等待成功,该方法是同步阻塞的,绑定成功后返回一个ChannelFuture
            ChannelFuture f = b.bind(port).sync();

            // 等待服务端监听端口关闭,阻塞,等待服务端链路关闭之后main函数才退出
            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 PojoServer().bind(port);
    }

}

PojoServerHandler.java

package cn.xpleaf.basic;

import java.util.List;

import cn.xpleaf.pojo.User;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class PojoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 注意msg为List,而不是User类型,这点尤其需要注意
        // 否则程序人执行,不会报错,但没有任何输出
        @SuppressWarnings("unchecked")
        List<Object> list = (List<Object>) msg;
        System.out.println("Pojo from client : " + list);
        // 遍历List,输出的是pojo对象中的属性
        for (Object obj : list) {
            System.out.println(obj);
        }
    }

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

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

客户端

PojoClient.java

package cn.xpleaf.basic;

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 PojoClient {

    public void connect(int port, String host) 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 MsgpackDecoder());
                        // 添加MessagePack编码器
                        ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder());
                        // 添加业务处理handler
                        ch.pipeline().addLast(new PojoClientHandler());
                    }
                });
            // 发起异步连接操作(注意服务端是bind,客户端则需要connect)
            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) {
                // 采用默认值
            }
        }
        new PojoClient().connect(port, "localhost");
    }
}

PojoClientHandler.java

package cn.xpleaf.basic;

import cn.xpleaf.pojo.User;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class PojoClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        User user = new User();
        user.setName("client");
        user.setAge(10);
//      for(int i = 0; i < 10; i++) {
//          ctx.write(user);
//      }
//      ctx.flush();
        ctx.writeAndFlush(user);
    }

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

    }

    @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 + "]";
    }

}

测试

运行服务端,再运行客户端,服务端的输出结果如下:

Pojo from client : ["client",10]
"client"
10

MessagePack在Netty中的应用

标签:泛型类   time   缓冲   cli   程序   cep   his   t对象   学习   

原文:http://blog.51cto.com/xpleaf/2071686

(0)
(0)
   
举报
评论 一句话评论(0
0条  
登录后才能评论!
© 2014 bubuko.com 版权所有 鲁ICP备09046678号-4
打开技术之扣,分享程序人生!
             

鲁公网安备 37021202000002号