netty学习之路四，利用Marshalling序列化对象，传输

深碍√TFBOYSˉ_ 2022-03-09 08:18 298阅读 0赞

### 利用Marshalling序列化对象 ###

*  pom依赖
 *  Server
 *  ServerHandler
 *  Client
 *  ClientHandler
 *  实体类
 *   *  response和request
 *  Marshalling工厂类

# pom依赖 #

<dependency>
               <groupId>io.netty</groupId>
               <artifactId>netty-all</artifactId>
               <version>5.0.0.Alpha2</version>
           </dependency>
    
           <dependency>
               <groupId>org.jboss.marshalling</groupId>
               <artifactId>jboss-marshalling-serial</artifactId>
               <version>2.0.0.Beta2</version>
           </dependency>

# Server #

public class Server {
        public static void main(String[] args) throws Exception{
    
            EventLoopGroup bossGroup = new NioEventLoopGroup(); // 负责处理来自客户端的链接
            EventLoopGroup workerGroup = new NioEventLoopGroup();//负责处理任务
    
            ServerBootstrap bootstrap = new ServerBootstrap();//配置类
            bootstrap.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());//添加处理器
                            socketChannel.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());//添加处理器
                            socketChannel.pipeline().addLast(new ServerHandler());//添加处理器
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
    
            ChannelFuture future = bootstrap.bind(8888).sync();
            future.channel().closeFuture().sync();//阻塞线程
    
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
    
        }
    }

# ServerHandler #

public class ServerHandler extends ChannelHandlerAdapter {
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("channelActive is invoked");
        }
    
        /**
         * 收到数据时调用
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            try{
                Request request = (Request) msg;
                System.out.println(request);
                Response response = new Response();
                response.setAge(request.getAge());
                response.setName(request.getName());
                ctx.writeAndFlush(response);
            }finally {
                ReferenceCountUtil.release(msg);
            }
        }
    
        /**
         * 抛出异常时调用
         * @param ctx
         * @param cause
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

# Client #

public class Client {
        public static void main(String[] args) throws Exception{
            EventLoopGroup workerGroup = new NioEventLoopGroup();//负责处理任务
    
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());//添加处理器
                            socketChannel.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());//添加处理器
                            socketChannel.pipeline().addLast(new ClientHandler());
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .option(ChannelOption.SO_KEEPALIVE, true);
    
            Request request = new Request();
            request.setName("xxxx");
            request.setAge(20);
            ChannelFuture f = bootstrap.connect("127.0.0.1",8888).sync();
            f.channel().writeAndFlush(request);
    
    
            f.channel().closeFuture().sync();
            workerGroup.shutdownGracefully();
    
        }
    }

# ClientHandler #

public class ClientHandler extends ChannelHandlerAdapter {
        @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 {
            cause.printStackTrace();
            ctx.close();
        }
    
    }

# 实体类 #

## response和request ##

public class Request implements Serializable {
    
        private String name;
        private int age;
    
        public String getName() {
            return name;
        }
    
        @Override
        public String toString() {
            return "Request{" +
                    "name='" + name + '\'' +
                    ", age=" + age +
                    '}';
        }
    
        public void setName(String name) {
            this.name = name;
        }
    
        public int getAge() {
            return age;
        }
    
        public void setAge(int age) {
            this.age = age;
        }
    }

# Marshalling工厂类 #

public 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;
        }
    }