Netty之传输POJO(使用JBoss的Marshalling序列化方式)

1、关于JBOSS的Marshalling的简介

       JBoss Marshalling 是一个Java对象的序列化API包，修正了JDK自带的序列化包的很多问题，但又保持跟
java.io.Serializable 接口的兼容；同时增加了一些可调的参数和附加的特性，而这些参数和特性可通过工厂类进行配置。

2、关于JBoss的Marshalling的配置

      2.1 服务端的配置

       



      2.2 客户端的配置

       



        2.3  直接发送和接收对象

          


3、关于JBoss的Marshalling的工厂类

import io.netty.handler.codec.marshalling.DefaultMarshallerProvider;
import io.netty.handler.codec.marshalling.DefaultUnmarshallerProvider;
import io.netty.handler.codec.marshalling.MarshallerProvider;
import io.netty.handler.codec.marshalling.MarshallingDecoder;
import io.netty.handler.codec.marshalling.MarshallingEncoder;
import io.netty.handler.codec.marshalling.UnmarshallerProvider;

import org.jboss.marshalling.MarshallerFactory;
import org.jboss.marshalling.Marshalling;
import org.jboss.marshalling.MarshallingConfiguration;

/**
* Marshalling工厂
*
*/
public final class MarshallingCodeCFactory {

/**
* 创建Jboss Marshalling解码器MarshallingDecoder
*
* @return MarshallingDecoder
*/
public static MarshallingDecoder buildMarshallingDecoder() {
// 首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。
final MarshallerFactory marshallerFactory = Marshalling
.getProvidedMarshallerFactory("serial");
// 创建了MarshallingConfiguration对象，配置了版本号为5
final MarshallingConfiguration configuration = new MarshallingConfiguration();
configuration.setVersion(5);
// 根据marshallerFactory和configuration创建provider
UnmarshallerProvider provider = new DefaultUnmarshallerProvider(
marshallerFactory, configuration);
// 构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度
MarshallingDecoder decoder = new MarshallingDecoder(provider,
1024 * 1024 * 1);
return decoder;
}

/**
* 创建Jboss Marshalling编码器MarshallingEncoder
*
* @return MarshallingEncoder
*/
public static MarshallingEncoder buildMarshallingEncoder() {
final MarshallerFactory marshallerFactory = Marshalling
.getProvidedMarshallerFactory("serial");
final MarshallingConfiguration configuration = new MarshallingConfiguration();
configuration.setVersion(5);
MarshallerProvider provider = new DefaultMarshallerProvider(
marshallerFactory, configuration);
// 构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组
MarshallingEncoder encoder = new MarshallingEncoder(provider);
return encoder;
}
}
4、服务端的实现
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;

public class Server {

public Server() {
}

public void bind(int port) throws Exception {
// 配置NIO线程组
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
// 服务器辅助启动类配置
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChildChannelHandler())//
.option(ChannelOption.SO_BACKLOG, 1024) // 设置tcp缓冲区 // (5)
.childOption(ChannelOption.SO_KEEPALIVE, true); // (6)
// 绑定端口 同步等待绑定成功
ChannelFuture f = b.bind(port).sync(); // (7)
// 等到服务端监听端口关闭
f.channel().closeFuture().sync();
} finally {
// 优雅释放线程资源
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}

/**
* 网络事件处理器
*/
private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 添加Jboss的序列化，编解码工具
ch.pipeline().addLast(
MarshallingCodeCFactory.buildMarshallingEncoder());
ch.pipeline().addLast(
MarshallingCodeCFactory.buildMarshallingDecoder());
// 处理网络IO
ch.pipeline().addLast(new ServerHandler());
}
}

public static void main(String[] args) throws Exception {
new Server().bind(9999);
}
}
5、服务端的Handler的实现
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

public class ServerHandler extends ChannelHandlerAdapter {
// 用于获取客户端发送的信息
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
// 用于获取客户端发来的数据信息
Request body = (Request) msg;
System.out.println("Server接受的客户端的信息 :" + body.toString());

// 会写数据给客户端
Response response = new Response(Integer.parseInt(body.getUrl()),
"xiaoming");
// 当服务端完成写操作后，关闭与客户端的连接
ctx.writeAndFlush(response);
// .addListener(ChannelFutureListener.CLOSE);

// 当有写操作时，不需要手动释放msg的引用
// 当只有读操作时，才需要手动释放msg的引用
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
// cause.printStackTrace();
ctx.close();
}
}6、客户端的实现
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 Client {

/**
* 连接服务器
*
* @param port
* @param host
* @throws Exception
*/
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)//
.handler(new MyChannelHandler());//
// 异步链接服务器 同步等待链接成功
ChannelFuture f = b.connect(host, port).sync();

// 等待链接关闭
f.channel().closeFuture().sync();

} finally {
group.shutdownGracefully();
System.out.println("客户端优雅的释放了线程资源...");
}

}

/**
* 网络事件处理器
*/
private class MyChannelHandler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 添加Jboss的序列化，编解码工具
ch.pipeline().addLast(
MarshallingCodeCFactory.buildMarshallingEncoder());
ch.pipeline().addLast(
MarshallingCodeCFactory.buildMarshallingDecoder());
// 处理网络IO
ch.pipeline().addLast(new ClientHandler());// 处理网络IO
}

}

public static void main(String[] args) throws Exception {
new Client().connect(9999, "127.0.0.1");

}

}
7、客户端的Handler的实现
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;

//用于读取客户端发来的信息
public class ClientHandler extends ChannelHandlerAdapter {

// 客户端与服务端，连接成功的售后
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 发送消息
Request request1 = new Request("666");
Request request2 = new Request("777");
Request request3 = new Request("888");
ctx.writeAndFlush(request1);
ctx.writeAndFlush(request2);
Thread.sleep(2000);
ctx.writeAndFlush(request3);
}

// 只是读数据，没有写数据的话
// 需要自己手动的释放的消息
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
try {
Response response = (Response) msg;
System.out.println(response);

} finally {
ReferenceCountUtil.release(msg);
}
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
ctx.close();
}

}
8、Request请求信息的实现
import java.io.Serializable;

public class Request implements Serializable {
/**
*
*/
private static final long serialVersionUID = -7033707301911915196L;
private String url;

public Request() {
}

public Request(String url) {
this.url = url;
}

public String getUrl() {
return url;
}

public void setUrl(String url) {
this.url = url;
}

@Override
public String toString() {
return "Request [url=" + url + "]";
}

}
9、Response请求信息的实现
public class Response implements Serializable {
/**
*
*/
private static final long serialVersionUID = -6236340795725143988L;
private int age;
private String name;

public Response() {
}

public Response(int age, String name) {
this.age = age;
this.name = name;
}

public int getAge() {
return age;
}

public void setAge(int age) {
this.age = age;
}

public String getName() {
return name;
}

public void setName(String name) {
this.name = name;
}

@Override
public String toString() {
return "Response [age=" + age + ", name=" + name + "]";
}

}