架构师养成记--21.netty编码解码

背景

作为网络传输框架，免不了哟啊传输对象，对象在传输之前就要序列化，这个序列化的过程就是编码过程。接收到编码后的数据就需要解码，还原传输的数据。

代码

工厂类

 1 import io.netty.handler.codec.marshalling.DefaultMarshallerProvider;
 2 import io.netty.handler.codec.marshalling.DefaultUnmarshallerProvider;
 3 import io.netty.handler.codec.marshalling.MarshallerProvider;
 4 import io.netty.handler.codec.marshalling.MarshallingDecoder;
 5 import io.netty.handler.codec.marshalling.MarshallingEncoder;
 6 import io.netty.handler.codec.marshalling.UnmarshallerProvider;
 7 
 8 import org.jboss.marshalling.MarshallerFactory;
 9 import org.jboss.marshalling.Marshalling;
10 import org.jboss.marshalling.MarshallingConfiguration;
11 
12 /**
13  * Marshalling工厂
14  * @author（alienware）
15  * @since 2014-12-16
16  */
17 public final class MarshallingCodeCFactory {
18 
19     /**
20      * 创建Jboss Marshalling解码器MarshallingDecoder
21      * @return MarshallingDecoder
22      */
23     public static MarshallingDecoder buildMarshallingDecoder() {
24         //首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。
25         final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
26         //创建了MarshallingConfiguration对象，配置了版本号为5 
27         final MarshallingConfiguration configuration = new MarshallingConfiguration();
28         configuration.setVersion(5);
29         //根据marshallerFactory和configuration创建provider
30         UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);
31         //构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度
32         MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024 * 1024 * 1);
33         return decoder;
34     }
35 
36     /**
37      * 创建Jboss Marshalling编码器MarshallingEncoder
38      * @return MarshallingEncoder
39      */
40     public static MarshallingEncoder buildMarshallingEncoder() {
41         final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
42         final MarshallingConfiguration configuration = new MarshallingConfiguration();
43         configuration.setVersion(5);
44         MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration);
45         //构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组
46         MarshallingEncoder encoder = new MarshallingEncoder(provider);
47         return encoder;
48     }
49 }

服务器端

 1 import io.netty.bootstrap.ServerBootstrap;
 2 import io.netty.channel.ChannelFuture;
 3 import io.netty.channel.ChannelInitializer;
 4 import io.netty.channel.ChannelOption;
 5 import io.netty.channel.EventLoopGroup;
 6 import io.netty.channel.nio.NioEventLoopGroup;
 7 import io.netty.channel.socket.SocketChannel;
 8 import io.netty.channel.socket.nio.NioServerSocketChannel;
 9 import io.netty.handler.logging.LogLevel;
10 import io.netty.handler.logging.LoggingHandler;
11 
12 public class Server {
13 
14     public static void main(String[] args) throws Exception{
15         
16         EventLoopGroup pGroup = new NioEventLoopGroup();
17         EventLoopGroup cGroup = new NioEventLoopGroup();
18         
19         ServerBootstrap b = new ServerBootstrap();
20         b.group(pGroup, cGroup)
21          .channel(NioServerSocketChannel.class)
22          .option(ChannelOption.SO_BACKLOG, 1024)
23          //设置日志
24          .handler(new LoggingHandler(LogLevel.INFO))
25          .childHandler(new ChannelInitializer<SocketChannel>() {
26             protected void initChannel(SocketChannel sc) throws Exception {
27                 sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
28                 sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
29                 sc.pipeline().addLast(new ServerHandler());
30             }
31         });
32         
33         ChannelFuture cf = b.bind(8765).sync();
34         
35         cf.channel().closeFuture().sync();
36         pGroup.shutdownGracefully();
37         cGroup.shutdownGracefully();
38         
39     }
40 }

 1 import io.netty.channel.ChannelHandlerAdapter;
 2 import io.netty.channel.ChannelHandlerContext;
 3 
 4 import java.io.File;
 5 import java.io.FileOutputStream;
 6 
 7 import bhz.utils.GzipUtils;
 8 
 9 public class ServerHandler extends ChannelHandlerAdapter{
10 
11     @Override
12     public void channelActive(ChannelHandlerContext ctx) throws Exception {
13 
14     }
15 
16     @Override
17     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
18         Req req = (Req)msg;
19         System.out.println("Server : " + req.getId() + ", " + req.getName() + ", " + req.getRequestMessage());
20         byte[] attachment = GzipUtils.ungzip(req.getAttachment());
21         
22         String path = System.getProperty("user.dir") + File.separatorChar + "receive" +  File.separatorChar + "001.jpg";
23         FileOutputStream fos = new FileOutputStream(path);
24         fos.write(attachment);
25         fos.close();
26         
27         Resp resp = new Resp();
28         resp.setId(req.getId());
29         resp.setName("resp" + req.getId());
30         resp.setResponseMessage("响应内容" + req.getId());
31         ctx.writeAndFlush(resp);//.addListener(ChannelFutureListener.CLOSE);
32     }
33 
34     @Override
35     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
36         
37     }
38 
39     @Override
40     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
41         ctx.close();
42     }
43 
44     
45     
46 }

客户端

 1 import io.netty.bootstrap.Bootstrap;
 2 import io.netty.channel.ChannelFuture;
 3 import io.netty.channel.ChannelInitializer;
 4 import io.netty.channel.EventLoopGroup;
 5 import io.netty.channel.nio.NioEventLoopGroup;
 6 import io.netty.channel.socket.SocketChannel;
 7 import io.netty.channel.socket.nio.NioSocketChannel;
 8 
 9 import java.io.File;
10 import java.io.FileInputStream;
11 
12 import bhz.utils.GzipUtils;
13 
14 public class Client {
15 
16     
17     public static void main(String[] args) throws Exception{
18         
19         EventLoopGroup group = new NioEventLoopGroup();
20         Bootstrap b = new Bootstrap();
21         b.group(group)
22          .channel(NioSocketChannel.class)
23          .handler(new ChannelInitializer<SocketChannel>() {
24             @Override
25             protected void initChannel(SocketChannel sc) throws Exception {
26                 sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
27                 sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
28                 sc.pipeline().addLast(new ClientHandler());
29             }
30         });
31         
32         ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();
33         
34         for(int i = 0; i < 5; i++ ){
35             Req req = new Req();
36             req.setId("" + i);
37             req.setName("pro" + i);
38             req.setRequestMessage("数据信息" + i);    
39             String path = System.getProperty("user.dir") + File.separatorChar + "sources" +  File.separatorChar + "001.jpg";
40             File file = new File(path);
41             FileInputStream in = new FileInputStream(file);  
42             byte[] data = new byte[in.available()];  
43             in.read(data);  
44             in.close(); 
45             req.setAttachment(GzipUtils.gzip(data));
46             cf.channel().writeAndFlush(req);
47         }
48 
49         cf.channel().closeFuture().sync();
50         group.shutdownGracefully();
51     }
52 }

 1 import io.netty.channel.ChannelHandlerAdapter;
 2 import io.netty.channel.ChannelHandlerContext;
 3 import io.netty.util.ReferenceCountUtil;
 4 
 5 public class ClientHandler extends ChannelHandlerAdapter{
 6     
 7     @Override
 8     public void channelActive(ChannelHandlerContext ctx) throws Exception {
 9         
10     }
11 
12     @Override
13     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
14         try {
15             Resp resp = (Resp)msg;
16             System.out.println("Client : " + resp.getId() + ", " + resp.getName() + ", " + resp.getResponseMessage());            
17         } finally {
18             ReferenceCountUtil.release(msg);
19         }
20     }
21 
22     @Override
23     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
24         
25     }
26 
27     @Override
28     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
29         ctx.close();
30     }
31     
32 }

工具类（压缩文件）

 1 import java.io.ByteArrayInputStream;
 2 import java.io.ByteArrayOutputStream;
 3 import java.io.File;
 4 import java.io.FileInputStream;
 5 import java.io.FileOutputStream;
 6 import java.util.zip.GZIPInputStream;
 7 import java.util.zip.GZIPOutputStream;
 8 
 9 public class GzipUtils {
10 
11     public static byte[] gzip(byte[] data) throws Exception{
12         ByteArrayOutputStream bos = new ByteArrayOutputStream();
13         GZIPOutputStream gzip = new GZIPOutputStream(bos);
14         gzip.write(data);
15         gzip.finish();
16         gzip.close();
17         byte[] ret = bos.toByteArray();
18         bos.close();
19         return ret;
20     }
21     
22     public static byte[] ungzip(byte[] data) throws Exception{
23         ByteArrayInputStream bis = new ByteArrayInputStream(data);
24         GZIPInputStream gzip = new GZIPInputStream(bis);
25         byte[] buf = new byte[1024];
26         int num = -1;
27         ByteArrayOutputStream bos = new ByteArrayOutputStream();
28         while((num = gzip.read(buf, 0 , buf.length)) != -1 ){
29             bos.write(buf, 0, num);
30         }
31         gzip.close();
32         bis.close();
33         byte[] ret = bos.toByteArray();
34         bos.flush();
35         bos.close();
36         return ret;
37     }
38     
39     public static void main(String[] args) throws Exception{
40         
41         //读取文件
42         String readPath = System.getProperty("user.dir") + File.separatorChar + "sources" +  File.separatorChar + "006.jpg";
43         File file = new File(readPath);  
44         FileInputStream in = new FileInputStream(file);  
45         byte[] data = new byte[in.available()];  
46         in.read(data);  
47         in.close();  
48         
49         System.out.println("文件原始大小:" + data.length);
50         //测试压缩
51         
52         byte[] ret1 = GzipUtils.gzip(data);
53         System.out.println("压缩之后大小:" + ret1.length);
54         
55         byte[] ret2 = GzipUtils.ungzip(ret1);
56         System.out.println("还原之后大小:" + ret2.length);
57         
58         //写出文件
59         String writePath = System.getProperty("user.dir") + File.separatorChar + "receive" +  File.separatorChar + "006.jpg";
60         FileOutputStream fos = new FileOutputStream(writePath);
61         fos.write(ret2);
62         fos.close();        
63         
64         
65     }
66     
67     
68     
69 }

数据对象

 1 import java.io.Serializable;
 2 
 3 public class Req implements Serializable{
 4 
 5     private static final long  SerialVersionUID = 1L;
 6     
 7     private String id ;
 8     private String name ;
 9     private String requestMessage ;
10     private byte[] attachment;
11     
12     public String getId() {
13         return id;
14     }
15     public void setId(String id) {
16         this.id = id;
17     }
18     public String getName() {
19         return name;
20     }
21     public void setName(String name) {
22         this.name = name;
23     }
24     public String getRequestMessage() {
25         return requestMessage;
26     }
27     public void setRequestMessage(String requestMessage) {
28         this.requestMessage = requestMessage;
29     }
30     public byte[] getAttachment() {
31         return attachment;
32     }
33     public void setAttachment(byte[] attachment) {
34         this.attachment = attachment;
35     }
36     
37     
38 
39 
40 }

 1 import java.io.Serializable;
 2 
 3 public class Resp implements Serializable{
 4     
 5     private static final long serialVersionUID = 1L;
 6     
 7     private String id;
 8     private String name;
 9     private String responseMessage;
10     
11     public String getId() {
12         return id;
13     }
14     public void setId(String id) {
15         this.id = id;
16     }
17     public String getName() {
18         return name;
19     }
20     public void setName(String name) {
21         this.name = name;
22     }
23     public String getResponseMessage() {
24         return responseMessage;
25     }
26     public void setResponseMessage(String responseMessage) {
27         this.responseMessage = responseMessage;
28     }
29     
30 
31 }