使用JDK的ObjectInputStream和ObjectOutputStream能够实现java对象的序列化和反序列化(只要被序列化的POJO对象实现Serializable接口)。java
在不须要考虑跨语言,而且对序列化的性能要去不苛刻时,JDK默认的序列化机制是最明智的选择之一。web
下面的例子中,咱们使用Netty的ObjectDecoder和ObjectEncoder对请求和应答对象进行序列化。
请求对象:编程
/** * @author j.tommy * @version 1.0 * @date 2017/11/18 */
public class SubscribeReq implements Serializable {
private int subReqId;
private String userName;
private String productName;
public SubscribeReq() {
}
public SubscribeReq(int subReqId, String userName, String productName) {
this.subReqId = subReqId;
this.userName = userName;
this.productName = productName;
}
public int getSubReqId() {
return subReqId;
}
public void setSubReqId(int subReqId) {
this.subReqId = subReqId;
}
public String getUserName() {
return userName;
}
public void setUserName(String userName) {
this.userName = userName;
}
public String getProductName() {
return productName;
}
public void setProductName(String productName) {
this.productName = productName;
}
@Override
public String toString() {
return "SubscribeReq{" +
"subReqId=" + subReqId +
", userName='" + userName + '\'' +
", productName='" + productName + '\'' +
'}';
}
}
应答对象:缓存
/** * @author j.tommy * @version 1.0 * @date 2017/11/18 */
public class SubscribeResp implements Serializable {
private int subReqId;
private int respCode;
private String desc;
public SubscribeResp() {
}
public SubscribeResp(int subReqId, int respCode) {
this.subReqId = subReqId;
this.respCode = respCode;
}
public SubscribeResp(int subReqId, int respCode, String desc) {
this.subReqId = subReqId;
this.respCode = respCode;
this.desc = desc;
}
@Override
public String toString() {
return "SubscribeResp{" +
"subReqId=" + subReqId +
", respCode=" + respCode +
", desc='" + desc + '\'' +
'}';
}
}
服务端
在ChannelPipeline中增长ObjectDecoder解码器和ObjectEncoder编码器,并对要进行java序列化的对象实现Serialzable接口。安全
/** * @author j.tommy * @version 1.0 * @date 2017/11/18 */
public class SubReqServer {
public static void main(String[] args) {
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>() {
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new ObjectDecoder(1024*1024, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
socketChannel.pipeline().addLast(new ObjectEncoder());
socketChannel.pipeline().addLast(new SubReqServerHandler());
}
});
try {
ChannelFuture f = b.bind(9988).sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
class SubReqServerHandler extends ChannelHandlerAdapter {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
SubscribeReq subscribeReq = (SubscribeReq) msg;
System.out.println("接收到客户端请求:" + subscribeReq);
ctx.writeAndFlush(resp(subscribeReq.getSubReqId()));
}
private SubscribeResp resp(int subReqId) {
SubscribeResp subscribeResp = new SubscribeResp(subReqId,0);
return subscribeResp;
}
}
客户端多线程
/** * @author j.tommy * @version 1.0 * @date 2017/11/18 */
public class SubReqClient {
public static void main(String[] args) {
EventLoopGroup group = new NioEventLoopGroup();
Bootstrap b = new Bootstrap();
b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).handler(new ChannelInitializer<SocketChannel>() {
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new ObjectDecoder(1024*1024, ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
socketChannel.pipeline().addLast(new ObjectEncoder());
socketChannel.pipeline().addLast(new SubReqClientHandler());
}
});
try {
ChannelFuture f = b.connect("127.0.0.1",9988).sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
group.shutdownGracefully();
}
}
}
class SubReqClientHandler extends ChannelHandlerAdapter {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 链路激活后,向服务端发送10条订购信息
for (int i=0;i<10;i++) {
SubscribeReq subscribeReq = new SubscribeReq(i,"tommy","Netty权威指南");
ctx.write(subscribeReq);
}
ctx.flush();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
SubscribeResp subscribeResp = (SubscribeResp) msg;
System.out.println("接收到服务端响应:" + subscribeResp);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
}
运行结果:
服务端
客户端
并发
ObjectDecoder负责对实现了Serialzable接口的POJO对象进行解码,它有多个构造函数,支持不一样的ClassResolver。
服务端使用的是weakCachingConcurrentResolver,建立线程安全的WeakReferenceMap对类加载器进行缓存,它支持多线程并发访问。当虚拟机内存不足时,会释放缓存中的内存。
客户端使用的是cacheDisabled,禁止对类加载器进行缓存,它在基于OSGi的动态模块化编程中常用。因为OSGi的bundle能够进行热部署和热升级,当某个bundle升级后,它对应的类加载器也将一块儿升级,所以在动态模块化编程中,不多对类加载器进行缓存,由于它随时可能发生变化。
为了防止异常码流和解码错位致使的内存溢出,这里将单个对象的最大序列化后的字节数设置为1M。socket
ObjectEncoder负责将实现了Serialzable接口的POJO对象进行编码,用户对对象手动进行序列化,关注于本身的业务便可。
对象的序列化和反序列化都由Netty的ObjectDecoder和ObjectEncoder来搞定。ide
参考《Netty权威指南》svg