netty編解碼之jboss marshalling
netty編解碼之jboss marshalling
jboss marshalling是jboss內部的一個序列化框架,速度也十分快,這裡netty也提供了支援,使用十分方便,不需要像protobuf一樣編寫proto檔案,只需要提供兩個編解碼器即可,以下就是jboss marshalling使用的開始。
原始碼程式
model類
model類和之前編寫java序列化的時候沒有區別,這裡便不再多說,僅僅貼出原始碼:
SubscribeReq:
package cn.com.serialize;
import java.io.Serializable;
/**
* Created by xiaxuan on 17/11/27.
*/
public class SubscribeReq implements Serializable {
private int subReqID;
private String userName;
private String productName;
private String phoneNumber;
private String address;
//getter setter
...
@Override
public String toString() {
return "SubscribeReq{" +
"subReqID=" + subReqID +
", userName='" + userName + '\'' +
", productName='" + productName + '\'' +
", phoneNumber='" + phoneNumber + '\'' +
", address='" + address + '\'' +
'}';
}
}
SubscribeResp
package cn.com.serialize;
import java.io.Serializable;
/**
* Created by xiaxuan on 17/11/27.
*/
public class SubscribeResp implements Serializable {
private int subReqID;
private int respCode;
private String desc;
//getter setter
...
@Override
public String toString() {
return "SubscribeResp{" +
"subReqID=" + subReqID +
", respCode=" + respCode +
", desc='" + desc + '\'' +
'}';
}
}
marshalling編解碼工廠類
MarshallingCodeCFactory:
package cn.com.marshalling;
import io.netty.handler.codec.marshalling.*;
import org.jboss.marshalling.MarshallerFactory;
import org.jboss.marshalling.Marshalling;
import org.jboss.marshalling.MarshallingConfiguration;
/**
* Created by xiaxuan on 17/11/28.
*/
public final class MarshallingCodeCFactory {
public static MarshallingDecoder buildMarshallingDecoder() {
final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
final MarshallingConfiguration configuration = new MarshallingConfiguration();
configuration.setVersion(5);
UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);
MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024);
return decoder;
}
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;
}
}
這裡在獲取factory的時候傳入的引數為serial表示為建立java序列化工廠物件。
在建立MarshallingDecoder物件時傳入兩個引數,分別是UnmarshallerProvider和單個訊息序列化後的最大長度。
server程式
SubReqServer:
package cn.com.marshalling;
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;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
/**
* Created by xiaxuan on 17/11/28.
*/
public class SubReqServer {
public void bind(int port) {
//配置服務端NIO執行緒組
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup 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(
MarshallingCodeCFactory.buildMarshallingDecoder()
);
ch.pipeline().addLast(
MarshallingCodeCFactory.buildMarshallingEncoder()
);
ch.pipeline().addLast(new SubReqServerHandler());
}
});
//繫結埠,同步等待成功
ChannelFuture f = b.bind(port).sync();
//等待服務端監聽埠關閉
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
//優雅退出
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
int port = 8080;
new SubReqServer().bind(port);
}
}
這裡我們將先前MarshallingFactory工廠類建立的MarshallingDecoder解碼器和MarshallingEncoder編碼器加入到ChannelPipline中,來進行對我們的傳輸物件進行編解碼。
SubReqServerHandler:
package cn.com.marshalling;
import cn.com.serialize.SubscribeReq;
import cn.com.serialize.SubscribeResp;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
@Sharable
public class SubReqServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
SubscribeReq req = (SubscribeReq) msg;
if ("Xiaxuan".equalsIgnoreCase(req.getUserName())) {
System.out.println("Service accept client subscrib req : ["
+ req.toString() + "]");
ctx.writeAndFlush(resp(req.getSubReqID()));
}
}
private SubscribeResp resp(int subReqID) {
SubscribeResp resp = new SubscribeResp();
resp.setSubReqID(subReqID);
resp.setRespCode(0);
resp.setDesc("Netty book order succeed, 3 days later, sent to the designated address");
return resp;
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();// 發生異常,關閉鏈路
}
}
業務邏輯與之前並沒有太多不同,均是獲取到客戶端請求後進行響應。
client程式
SubReqClient:
package cn.com.marshalling;
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;
/**
* Created by xiaxuan on 17/12/5.
*/
public class SubReqClient {
public void connect(int port, String host) {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
ch.pipeline().addLast(new SubReqClientHandler());
}
});
//發起非同步連線操作
ChannelFuture f = b.connect(host, port).sync();
//等待客戶端鏈路關閉
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) {
int port = 8080;
String host = "127.0.0.1";
new SubReqClient().connect(port, host);
}
}
除了編解碼框架以外,其他的並沒有不同的。
SubReqClientHandler:
package cn.com.marshalling;
import cn.com.serialize.SubscribeReq;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
public class SubReqClientHandler extends ChannelInboundHandlerAdapter {
public SubReqClientHandler() {
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
for (int i = 0; i < 10; i++) {
ctx.write(subReq(i));
}
ctx.flush();
}
private SubscribeReq subReq(int i) {
SubscribeReq req = new SubscribeReq();
req.setAddress("NanJing YuHuaTai");
req.setPhoneNumber("135xxxxxxxx");
req.setProductName("Netty Book For Marshalling");
req.setSubReqID(i);
req.setUserName("Xiaxuan");
return req;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
System.out.println("Receive server response : [" + msg + "]");
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
客戶端業務邏輯與之前也沒有其他的區別,都是構建十次請求,然後一次性發出,最後輸出服務端的響應。
程式執行
分別啟動server程式和client程式,檢視執行結果,如下:
server:
client:
執行成功。
使用jboss marshalling來進行編解碼還是比較簡單的,在這裡我們模擬了TCP的粘包/拆包場景,但是程式的執行結果仍然正確,說明Marshalling的編解碼器支援半包和粘包的處理,對於普通的開發者來說,只需要將Marshalling編碼器和解碼器加入到ChannelPipline中,就能實現對Marshalling序列化的支援。