MessagePack在Netty中的應用
阿新 • • 發佈：2018-02-15
泛型類 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"
10MessagePack在Netty中的應用