jboss marshell序列化整合netty

ゝ一纸荒年。 2022-06-08 04:41 178阅读 0赞

1、关于JBOSS的Marshalling的简介

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

2、关于JBoss的Marshalling的配置

2.1 服务端的配置

![20161215171510377][]

2.2 客户端的配置

![20161215171522490][]

2.3  直接发送和接收对象

![20161215171859831][]

3、关于JBoss的Marshalling的工厂类

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

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.  \*  
15.  \*/  
16. **public** **final** **class** MarshallingCodeCFactory \{  
17.   
18.     /\*\* 
19.      \* 创建Jboss Marshalling解码器MarshallingDecoder 
20.      \*  
21.      \* @return MarshallingDecoder 
22.      \*/  
23.     **public** **static** MarshallingDecoder buildMarshallingDecoder() \{  
24.         // 首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。  
25.         **final** MarshallerFactory marshallerFactory = Marshalling  
26.                 .getProvidedMarshallerFactory("serial");  
27.         // 创建了MarshallingConfiguration对象，配置了版本号为5  
28.         **final** MarshallingConfiguration configuration = **new** MarshallingConfiguration();  
29.         configuration.setVersion(5);  
30.         // 根据marshallerFactory和configuration创建provider  
31.         UnmarshallerProvider provider = **new** DefaultUnmarshallerProvider(  
32.                 marshallerFactory, configuration);  
33.         // 构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度  
34.         MarshallingDecoder decoder = **new** MarshallingDecoder(provider,  
35.                 1024 \* 1024 \* 1);  
36.         **return** decoder;  
37.     \}  
38.   
39.     /\*\* 
40.      \* 创建Jboss Marshalling编码器MarshallingEncoder 
41.      \*  
42.      \* @return MarshallingEncoder 
43.      \*/  
44.     **public** **static** MarshallingEncoder buildMarshallingEncoder() \{  
45.         **final** MarshallerFactory marshallerFactory = Marshalling  
46.                 .getProvidedMarshallerFactory("serial");  
47.         **final** MarshallingConfiguration configuration = **new** MarshallingConfiguration();  
48.         configuration.setVersion(5);  
49.         MarshallerProvider provider = **new** DefaultMarshallerProvider(  
50.                 marshallerFactory, configuration);  
51.         // 构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组  
52.         MarshallingEncoder encoder = **new** MarshallingEncoder(provider);  
53.         **return** encoder;  
54.     \}  
55. \}

4、服务端的实现

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

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** Server() \{  
15.     \}  
16.   
17.     **public** **void** bind(**int** port) **throws** Exception \{  
18.         // 配置NIO线程组  
19.         EventLoopGroup bossGroup = **new** NioEventLoopGroup();  
20.         EventLoopGroup workerGroup = **new** NioEventLoopGroup();  
21.         **try** \{  
22.             // 服务器辅助启动类配置  
23.             ServerBootstrap b = **new** ServerBootstrap();  
24.             b.group(bossGroup, workerGroup)  
25.                     .channel(NioServerSocketChannel.**class**)  
26.                     .handler(**new** LoggingHandler(LogLevel.INFO))  
27.                     .childHandler(**new** ChildChannelHandler())//  
28.                     .option(ChannelOption.SO\_BACKLOG, 1024) // 设置tcp缓冲区 // (5)  
29.                     .childOption(ChannelOption.SO\_KEEPALIVE, **true**); // (6)  
30.             // 绑定端口 同步等待绑定成功  
31.             ChannelFuture f = b.bind(port).sync(); // (7)  
32.             // 等到服务端监听端口关闭  
33.             f.channel().closeFuture().sync();  
34.         \} **finally** \{  
35.             // 优雅释放线程资源  
36.             workerGroup.shutdownGracefully();  
37.             bossGroup.shutdownGracefully();  
38.         \}  
39.     \}  
40.   
41.     /\*\* 
42.      \* 网络事件处理器 
43.      \*/  
44.     **private** **class** ChildChannelHandler **extends** ChannelInitializer<SocketChannel> \{  
45.         @Override  
46.         **protected** **void** initChannel(SocketChannel ch) **throws** Exception \{  
47.             // 添加Jboss的序列化，编解码工具  
48.             ch.pipeline().addLast(  
49.                     MarshallingCodeCFactory.buildMarshallingEncoder());  
50.             ch.pipeline().addLast(  
51.                     MarshallingCodeCFactory.buildMarshallingDecoder());  
52.             // 处理网络IO  
53.             ch.pipeline().addLast(**new** ServerHandler());  
54.         \}  
55.     \}  
56.   
57.     **public** **static** **void** main(String\[\] args) **throws** Exception \{  
58.         **new** Server().bind(9999);  
59.     \}  
60. \}

5、服务端的Handler的实现

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

1.  **import** io.netty.channel.ChannelHandlerAdapter;  
2.  **import** io.netty.channel.ChannelHandlerContext;  
3.    
4.  **public** **class** ServerHandler **extends** ChannelHandlerAdapter \{  
5.      // 用于获取客户端发送的信息  
6.      @Override  
7.      **public** **void** channelRead(ChannelHandlerContext ctx, Object msg)  
8.              **throws** Exception \{  
9.          // 用于获取客户端发来的数据信息  
10.         Request body = (Request) msg;  
11.         System.out.println("Server接受的客户端的信息 :" + body.toString());  
12.   
13.         // 会写数据给客户端  
14.         Response response = **new** Response(Integer.parseInt(body.getUrl()),  
15.                 "xiaoming");  
16.         // 当服务端完成写操作后，关闭与客户端的连接  
17.         ctx.writeAndFlush(response);  
18.         // .addListener(ChannelFutureListener.CLOSE);  
19.   
20.         // 当有写操作时，不需要手动释放msg的引用  
21.         // 当只有读操作时，才需要手动释放msg的引用  
22.     \}  
23.   
24.     @Override  
25.     **public** **void** exceptionCaught(ChannelHandlerContext ctx, Throwable cause)  
26.             **throws** Exception \{  
27.         // cause.printStackTrace();  
28.         ctx.close();  
29.     \}  
30. \}

6、客户端的实现

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

1.  **import** io.netty.bootstrap.Bootstrap;  
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.NioSocketChannel;  
9.    
10. **public** **class** Client \{  
11.   
12.     /\*\* 
13.      \* 连接服务器 
14.      \*  
15.      \* @param port 
16.      \* @param host 
17.      \* @throws Exception 
18.      \*/  
19.     **public** **void** connect(**int** port, String host) **throws** Exception \{  
20.         // 配置客户端NIO线程组  
21.         EventLoopGroup group = **new** NioEventLoopGroup();  
22.         **try** \{  
23.             // 客户端辅助启动类 对客户端配置  
24.             Bootstrap b = **new** Bootstrap();  
25.             b.group(group)//  
26.                     .channel(NioSocketChannel.**class**)//  
27.                     .option(ChannelOption.TCP\_NODELAY, **true**)//  
28.                     .handler(**new** MyChannelHandler());//  
29.             // 异步链接服务器 同步等待链接成功  
30.             ChannelFuture f = b.connect(host, port).sync();  
31.   
32.             // 等待链接关闭  
33.             f.channel().closeFuture().sync();  
34.   
35.         \} **finally** \{  
36.             group.shutdownGracefully();  
37.             System.out.println("客户端优雅的释放了线程资源...");  
38.         \}  
39.   
40.     \}  
41.   
42.     /\*\* 
43.      \* 网络事件处理器 
44.      \*/  
45.     **private** **class** MyChannelHandler **extends** ChannelInitializer<SocketChannel> \{  
46.         @Override  
47.         **protected** **void** initChannel(SocketChannel ch) **throws** Exception \{  
48.             // 添加Jboss的序列化，编解码工具  
49.             ch.pipeline().addLast(  
50.                     MarshallingCodeCFactory.buildMarshallingEncoder());  
51.             ch.pipeline().addLast(  
52.                     MarshallingCodeCFactory.buildMarshallingDecoder());  
53.             // 处理网络IO  
54.             ch.pipeline().addLast(**new** ClientHandler());// 处理网络IO  
55.         \}  
56.   
57.     \}  
58.   
59.     **public** **static** **void** main(String\[\] args) **throws** Exception \{  
60.         **new** Client().connect(9999, "127.0.0.1");  
61.   
62.     \}  
63.   
64. \}

7、客户端的Handler的实现

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

1.  **import** io.netty.channel.ChannelHandlerAdapter;  
2.  **import** io.netty.channel.ChannelHandlerContext;  
3.  **import** io.netty.util.ReferenceCountUtil;  
4.    
5.  //用于读取客户端发来的信息  
6.  **public** **class** ClientHandler **extends** ChannelHandlerAdapter \{  
7.    
8.      // 客户端与服务端，连接成功的售后  
9.      @Override  
10.     **public** **void** channelActive(ChannelHandlerContext ctx) **throws** Exception \{  
11.         // 发送消息  
12.         Request request1 = **new** Request("666");  
13.         Request request2 = **new** Request("777");  
14.         Request request3 = **new** Request("888");  
15.         ctx.writeAndFlush(request1);  
16.         ctx.writeAndFlush(request2);  
17.         Thread.sleep(2000);  
18.         ctx.writeAndFlush(request3);  
19.     \}  
20.   
21.     // 只是读数据，没有写数据的话  
22.     // 需要自己手动的释放的消息  
23.     @Override  
24.     **public** **void** channelRead(ChannelHandlerContext ctx, Object msg)  
25.             **throws** Exception \{  
26.         **try** \{  
27.             Response response = (Response) msg;  
28.             System.out.println(response);  
29.   
30.         \} **finally** \{  
31.             ReferenceCountUtil.release(msg);  
32.         \}  
33.     \}  
34.   
35.     @Override  
36.     **public** **void** exceptionCaught(ChannelHandlerContext ctx, Throwable cause)  
37.             **throws** Exception \{  
38.         ctx.close();  
39.     \}  
40.   
41. \}

8、 Request请求信息的实现

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

1.  **import** java.io.Serializable;  
2.    
3.  **public** **class** Request **implements** Serializable \{  
4.      /\*\* 
5.       \*  
6.       \*/  
7.      **private** **static** **final** **long** serialVersionUID = -7033707301911915196L;  
8.      **private** String url;  
9.    
10.     **public** Request() \{  
11.     \}  
12.   
13.     **public** Request(String url) \{  
14.         **this**.url = url;  
15.     \}  
16.   
17.     **public** String getUrl() \{  
18.         **return** url;  
19.     \}  
20.   
21.     **public** **void** setUrl(String url) \{  
22.         **this**.url = url;  
23.     \}  
24.   
25.     @Override  
26.     **public** String toString() \{  
27.         **return** "Request \[url=" + url + "\]";  
28.     \}  
29.   
30. \}

9、 Response请求信息的实现

**\[java\]**  [view plain][]   [copy][view plain]

[print][view plain] [?][view plain]

1.  **public** **class** Response **implements** Serializable \{  
2.      /\*\* 
3.       \*  
4.       \*/  
5.      **private** **static** **final** **long** serialVersionUID = -6236340795725143988L;  
6.      **private** **int** age;  
7.      **private** String name;  
8.    
9.      **public** Response() \{  
10.     \}  
11.   
12.     **public** Response(**int** age, String name) \{  
13.         **this**.age = age;  
14.         **this**.name = name;  
15.     \}  
16.   
17.     **public** **int** getAge() \{  
18.         **return** age;  
19.     \}  
20.   
21.     **public** **void** setAge(**int** age) \{  
22.         **this**.age = age;  
23.     \}  
24.   
25.     **public** String getName() \{  
26.         **return** name;  
27.     \}  
28.   
29.     **public** **void** setName(String name) \{  
30.         **this**.name = name;  
31.     \}  
32.   
33.     @Override  
34.     **public** String toString() \{  
35.         **return** "Response \[age=" + age + ", name=" + name + "\]";  
36.     \}  
37.   
38. \}

[20161215171510377]: /images/20220608/a0bfc5ecdbc14da9afb886c02f8525f1.png
[20161215171522490]: /images/20220608/10e23f9fde5947408d8b222bdeb1ee40.png
[20161215171859831]: /images/20220608/6b0daeabe8ad4fd08a3d6441735a9092.png
[view plain]: http://blog.csdn.net/zbw18297786698/article/details/53672802#