Netty之编解码
内容目录
MessagePack 简介MessagePack SDKMessagePack编码器开发MessagePack解码器编写客户端代码服务端代码POJO测试结果
MessagePack 简介
MessagePack是一个高效的二进制序列化和反序列化框架。
- 跨语言数据交换
- 性能更快
- 产生的码流更小
MessagePack SDK
<dependency>
<groupId>org.msgpack</groupId>
<artifactId>msgpack</artifactId>
<version>0.6.12</version>
</dependency>
MessagePack编码器开发
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);
}
}
MessagePack解码器编写
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));
}
}
客户端代码
public class EchoClient {
public void connect(String host, int port, int sendNumber) throws Exception {
// 配置客户端NIO线程组
try(EventLoopGroup group = new NioEventLoopGroup();) {
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 MsgpackDecoder());
// 添加长度字段编码器
// 在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();
}
}
public static void main(String[] args) throws Exception {
int port = 8888;
if(args != null && args.length > 0) {
try {
port = Integer.valueOf(port);
} catch (NumberFormatException e) {
// 采用默认值
}
}
int sendNumber = 100;
new EchoClient().connect("localhost", port, sendNumber);
}
}
public class EchoClientHandler extends ChannelHandlerAdapter {
// 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();
}
}
服务端代码
public class EchoServer {
public void bind(int port) throws Exception {
// 配置服务端的NIO线程组
try(EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup()) {
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 MsgpackDecoder());
// 添加长度字段编码器
// 在MessagePack编码器之前增加LengthFieldPrepender，它将在ByteBuf之前增加2个字节的消息长度字段
ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(2));
// 添加MessagePack编码器
ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder());
// 添加业务处理handler
// 添加业务处理handler
ch.pipeline().addLast(new EchoServerHandler());
}
});
// 绑定端口，同步等待成功
ChannelFuture f = b.bind(port).sync();
// 等待服务端监听端口关闭
f.channel().closeFuture().sync();
}
}
public static void main(String[] args) throws Exception {
int port = 8888;
if(args != null && args.length > 0) {
try {
port = Integer.valueOf(port);
} catch (NumberFormatException e) {
// TODO: handle exception
}
}
new EchoServer().bind(port);
}
}
public class EchoServerHandler extends ChannelHandlerAdapter {
@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();
}
}
POJO
@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 + "]";
}
}
测试结果
服务器端输出
Server receive the msgpack message : ["ABCDEFG --->0",0]
Server receive the msgpack message : ["ABCDEFG --->1",1]
Server receive the msgpack message : ["ABCDEFG --->2",2]
······省去代码······
Server receive the msgpack message : ["ABCDEFG --->98",98]
Server receive the msgpack message : ["ABCDEFG --->99",99]
客户端输出
Client receive the msgpack message : ["ABCDEFG --->0",0]
Client receive the msgpack message : ["ABCDEFG --->1",1]
Client receive the msgpack message : ["ABCDEFG --->2",2]
······省去代码······
Client receive the msgpack message : ["ABCDEFG --->98",98]
Client receive the msgpack message : ["ABCDEFG --->99",99]
本文参与 腾讯云自媒体同步曝光计划，分享自微信公众号。
原始发表：2018-09-13，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
相关产品与服务
文件存储
文件存储（Cloud File Storage，CFS）为您提供安全可靠、可扩展的共享文件存储服务。文件存储可与腾讯云服务器、容器服务、批量计算等服务搭配使用，为多个计算节点提供容量和性能可弹性扩展的高性能共享存储。腾讯云文件存储的管理界面简单、易使用，可实现对现有应用的无缝集成；按实际用量付费，为您节约成本，简化 IT 运维工作。