JBoss Marshalling序列化
Java是面向对象的开发语言。传递的数据如果是Java对象，应该是最方便且可靠。
/**
* 1. 双线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. 绑定服务监听端口并启动服务
*/
package com.bjsxt.socket.netty.serialized;
import com.bjsxt.socket.utils.SerializableFactory4Marshalling;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
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 Server4Serializable {
// 监听线程组，监听客户端请求
private EventLoopGroup acceptorGroup = null;
// 处理客户端相关操作线程组，负责处理与客户端的数据通讯
private EventLoopGroup clientGroup = null;
// 服务启动相关配置信息
private ServerBootstrap bootstrap = null;
public Server4Serializable(){
init();
}
private void init(){
acceptorGroup = new NioEventLoopGroup();
clientGroup = new NioEventLoopGroup();
bootstrap = new ServerBootstrap();
// 绑定线程组
bootstrap.group(acceptorGroup, clientGroup);
// 设定通讯模式为NIO
bootstrap.channel(NioServerSocketChannel.class);
// 设定缓冲区大小
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
// SO_SNDBUF发送缓冲区，SO_RCVBUF接收缓冲区，SO_KEEPALIVE开启心跳监测（保证连接有效）
bootstrap.option(ChannelOption.SO_SNDBUF, 16*1024)
.option(ChannelOption.SO_RCVBUF, 16*1024)
.option(ChannelOption.SO_KEEPALIVE, true);
}
public ChannelFuture doAccept(int port, final ChannelHandler... acceptorHandlers) throws InterruptedException{
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingDecoder());
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingEncoder());
ch.pipeline().addLast(acceptorHandlers);
}
});
ChannelFuture future = bootstrap.bind(port).sync();
return future;
}
public void release(){
this.acceptorGroup.shutdownGracefully();
this.clientGroup.shutdownGracefully();
}
public static void main(String[] args){
ChannelFuture future = null;
Server4Serializable server = null;
try{
server = new Server4Serializable();
future = server.doAccept(9999,new Server4SerializableHandler());
System.out.println("server started.");
future.channel().closeFuture().sync();
}catch(InterruptedException e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != server){
server.release();
}
}
}
}
/**
* @Sharable注解 -
* 代表当前Handler是一个可以分享的处理器。也就意味着，服务器注册此Handler后，可以分享给多个客户端同时使用。
* 如果不使用注解描述类型，则每次客户端请求时，必须为客户端重新创建一个新的Handler对象。
*
*/
package com.bjsxt.socket.netty.serialized;
import com.bjsxt.socket.utils.GzipUtils;
import com.bjsxt.socket.utils.RequestMessage;
import com.bjsxt.socket.utils.ResponseMessage;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
@Sharable
public class Server4SerializableHandler extends ChannelHandlerAdapter {
// 业务处理逻辑
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("from client : ClassName - " + msg.getClass().getName()
+ " ; message : " + msg.toString());
if(msg instanceof RequestMessage){
RequestMessage request = (RequestMessage)msg;
byte[] attachment = GzipUtils.unzip(request.getAttachment());
System.out.println(new String(attachment));
}
ResponseMessage response = new ResponseMessage(0L, "test response");
ctx.writeAndFlush(response);
}
// 异常处理逻辑
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("server exceptionCaught method run...");
cause.printStackTrace();
ctx.close();
}
}
/**
* 1. 单线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. connect连接服务，并发起请求
*/
package com.bjsxt.socket.netty.serialized;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import com.bjsxt.socket.utils.GzipUtils;
import com.bjsxt.socket.utils.RequestMessage;
import com.bjsxt.socket.utils.SerializableFactory4Marshalling;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
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 Client4Serializable {
// 处理请求和处理服务端响应的线程组
private EventLoopGroup group = null;
// 服务启动相关配置信息
private Bootstrap bootstrap = null;
public Client4Serializable(){
init();
}
private void init(){
group = new NioEventLoopGroup();
bootstrap = new Bootstrap();
// 绑定线程组
bootstrap.group(group);
// 设定通讯模式为NIO
bootstrap.channel(NioSocketChannel.class);
}
public ChannelFuture doRequest(String host, int port, final ChannelHandler... handlers) throws InterruptedException{
this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingDecoder());
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingEncoder());
ch.pipeline().addLast(handlers);
}
});
ChannelFuture future = this.bootstrap.connect(host, port).sync();
return future;
}
public void release(){
this.group.shutdownGracefully();
}
public static void main(String[] args) {
Client4Serializable client = null;
ChannelFuture future = null;
try{
client = new Client4Serializable();
future = client.doRequest("localhost", 9999, new Client4SerializableHandler());
String attachment = "test attachment";
byte[] attBuf = attachment.getBytes();
attBuf = GzipUtils.zip(attBuf);
RequestMessage msg = new RequestMessage(new Random().nextLong(),
"test", attBuf);
future.channel().writeAndFlush(msg);
TimeUnit.SECONDS.sleep(1);
future.addListener(ChannelFutureListener.CLOSE);
}catch(Exception e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != client){
client.release();
}
}
}
}
package com.bjsxt.socket.netty.serialized;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
public class Client4SerializableHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("from server : ClassName - " + msg.getClass().getName()
+ " ; message : " + msg.toString());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("client exceptionCaught method run...");
cause.printStackTrace();
ctx.close();
}
}
