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

电玩女神 2022-07-12 23:19 279阅读 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的工厂类

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 + "]";
    	}
    
    }

[20161215171510377]: /images/20220713/9dee8c484ce844aeb8cab27da95d184f.png
[20161215171522490]: /images/20220713/4fab00b70eed4dc98b8bc32ed9f08998.png
[20161215171859831]: /images/20220713/dbb63c543a9b4450a988fb40183b7fd2.png