Netty學習之路（九）-JBoss Marshalling編解碼
阿新 • • 發佈：2018-12-05
JBoss Marshalling 是一個Java物件序列化包，對JDK預設的序列化框架進行了優化，但又保持跟java.io.Serializable介面的相容，同時增加了一些可調的引數和附加的特性。
Marshalling開發環境準備
下載相關的Marshalling類庫:地址,將該jar匯入專案即可。
建立Marshalling編解碼器
通過建立MarshallingCodeCFactory工廠類來建立MarshallingDecoder解碼器與MarshallingEncoder編碼器。
package com.ph.Netty; import io.netty.handler.codec.marshalling.*; import org.jboss.marshalling.MarshallerFactory; import org.jboss.marshalling.Marshalling; import org.jboss.marshalling.MarshallingConfiguration; /** * Create by PH on 2018/11/10 */ public final class MarshallingCodeCFactory { /** * JBoss Marshalling 解碼器 * @return */ public static MarshallingDecoder buildMarshallingDecoder() { //引數“serial”表示建立的是Java序列化工廠物件 final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial"); final MarshallingConfiguration configuration = new MarshallingConfiguration(); configuration.setVersion(5); UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration); //1024表示單個訊息序列化後的最大長度 MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024); return decoder; } /** * JBoss Marshalling 編碼器 * @return */ public static MarshallingEncoder buildMarshallingEncoder() { final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial"); final MarshallingConfiguration configuration = new MarshallingConfiguration(); configuration.setVersion(5); MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration); MarshallingEncoder encoder = new MarshallingEncoder(provider); return encoder; } }
傳輸的POJO類
package com.ph.Netty; /** * Create by PH on 2018/11/10 */ public class BookInfo implements java.io.Serializable { private int id; private String name; private String type; public int getId() { return id; } public void setId(int id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getType() { return type; } public void setType(String type) { this.type = type; } @Override public String toString() { return "{" + "id=" + id + ", name=" + name + ", type='" + type + '\'' + '}'; } }
Netty的Marshalling服務端開發
package com.ph.Netty; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; /** * Create by PH on 2018/11/10 */ public class NettyServer { 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 NettyServer().bind(port); } public void bind(int port) throws Exception{ //NioEventLoopGroup是一個執行緒組，包含了一組NIO執行緒，專門用於網路事件的處理，實際上他們就是Reactor執行緒組 //bossGroup僅接收客戶端連線，不做複雜的邏輯處理，為了儘可能減少資源的佔用，取值越小越好 EventLoopGroup bossGroup = new NioEventLoopGroup(1); //用於進行SocketChannel的網路讀寫 EventLoopGroup workerGroup = new NioEventLoopGroup(); try { //是Netty用於啟動NIO服務端的輔助啟動類，目的是降低服務端的開發複雜度 ServerBootstrap b = new ServerBootstrap(); //配置NIO服務端 b.group(bossGroup, workerGroup) //指定使用NioServerSocketChannel產生一個Channel用來接收連線，他的功能對應於JDK // NIO類庫中的ServerSocketChannel類。 .channel(NioServerSocketChannel.class) //BACKLOG用於構造服務端套接字ServerSocket物件，標識當伺服器請求處理執行緒全滿時， // 用於臨時存放已完成三次握手的請求的佇列的最大長度。如果未設定或所設定的值小於1， // Java將使用預設值50。 .option(ChannelOption.SO_BACKLOG, 1024) //繫結I/O事件處理類，作用類似於Reactor模式中的Handler類，主要用於處理網路I/O事件 .childHandler(new ChannelInitializer<SocketChannel>() { protected void initChannel(SocketChannel arg0) throws Exception { arg0.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder()); arg0.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder()); arg0.pipeline().addLast(new ServerHandler()); } }); //繫結埠，同步等待繫結操作完成，完成後返回一個ChannelFuture,用於非同步操作的通知回撥 ChannelFuture f = b.bind(port).sync(); //等待服務端監聽埠關閉之後才退出main函式 f.channel().closeFuture().sync(); } finally { //退出，釋放執行緒池資源 bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } } /** * ChannelInboundHandlerAdapter實現自ChannelInboundHandler * ChannelInboundHandler提供了不同的事件處理方法你可以重寫 */ class ServerHandler extends ChannelInboundHandlerAdapter { /** * 接受客戶端傳送的訊息 * @param ctx * @param msg * @throws Exception */ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { BookInfo bookInfo = (BookInfo)msg; System.out.println("Server receive: " + bookInfo.toString()); BookInfo bookInfo1 = new BookInfo(); bookInfo1.setId(bookInfo.getId()); bookInfo1.setName("Server Netty Marshalling"); bookInfo1.setType("book order succeed"); ctx.writeAndFlush(bookInfo1); } public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { //將訊息傳送佇列中的訊息寫入到SocketChannel中傳送給對方 ctx.flush(); } public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { //當發生異常時釋放資源 ctx.close(); } }
Netty的Mashalling客戶端開發
package com.ph.Netty;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
/**
* Create by PH on 2018/11/10
*/
public class NettyClient {
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 NettyClient().connect(port, "127.0.0.1", 10);
}
public void connect(int port, String host, 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)
.handler(new ChannelInitializer<SocketChannel>() {
public void initChannel(SocketChannel ch) throws Exception{
ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
ch.pipeline().addLast(new ClientHandler(sendNumber));
}
});
//發起非同步連線操作
ChannelFuture f = b.connect(host, port).sync();
//等待客戶端鏈路關閉
f.channel().closeFuture().sync();
}finally {
group.shutdownGracefully();
}
}
}
class ClientHandler extends ChannelInboundHandlerAdapter {
private int sendNumber;
public ClientHandler(int sendNumber) {
this.sendNumber = sendNumber;
}
/**
* 當客戶端和服務端TCP鏈路建立成功之後，Netty的NIO執行緒會呼叫此方法
* @param ctx
*/
public void channelActive(ChannelHandlerContext ctx) {
for (int i=0;i<sendNumber;i++) {
BookInfo bookInfo = new BookInfo();
bookInfo.setId(i);
bookInfo.setName("Client Netty Marshalling");
bookInfo.setType("buy book");
ctx.write(bookInfo);
}
ctx.flush();
}
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception{
System.out.println("Client receive :" + msg);
}
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
ctx.close();
}
}
執行結果
服務端：
客戶端：
通過執行結果可看出並沒有發生粘包現象，說明Marshalling的編解碼器支援半包和粘包的處理，對於普通的開發者來說，只需要將Marshalling編碼器和解碼器加入到ChannelPipline中，就能實現對Marshalling序列化的支援。