JBoss Marshalling编解码

待我称王封你为后i 2022-03-28 11:39 288阅读 0赞

一、简介

JBoss Marshalling是一个Java对象序列化包，对JDK默认的序列化框架进行了优化，但又保持和java.io.Serializable接口的兼容。

二、Marshalling开发环境准备

1、下载类库

由于只涉及到Marshalling的序列化类库，因此只需要下载[jboss-marshalling-1.3.0.CR9.jar][]和[jboss-marshalling-serial-1.3.0.CR9.jar][]包即可，下载地址如下：[https://www.jboss.org/jbossmarshalling/downloads][https_www.jboss.org_jbossmarshalling_downloads]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70][]

2、将下载的类库build到classpath下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 1][]

二、Netty的Marshalling开发

1、Marshalling编解码器工厂类：

import org.jboss.marshalling.MarshallerFactory;
    import org.jboss.marshalling.Marshalling;
    import org.jboss.marshalling.MarshallingConfiguration;
    
    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;
    
    //Marshalling编解码器的工厂类
    public final class MarshallingCodeCFactory
    {
        //创建JBoss Marshalling解码器
        public static MarshallingDecoder buildMarshallingDecoder(){
            final MarshallerFactory marshallerFactory=Marshalling
                .getProvidedMarshallerFactory("serial");
            final MarshallingConfiguration configuration=new 
                MarshallingConfiguration();
            configuration.setVersion(5);
            UnmarshallerProvider provider=new DefaultUnmarshallerProvider(
                marshallerFactory, configuration);
            MarshallingDecoder decoder=new MarshallingDecoder(provider);
            return decoder;
        }
        
      //创建JBoss Marshalling编码器
        public static MarshallingEncoder buildMarshallingeEncoder(){
            //"serial"表示创建的是Java序列化工厂对象
            final MarshallerFactory marshallerFactory=Marshalling
                .getProvidedMarshallerFactory("serial");
            final MarshallingConfiguration configuration=new 
                MarshallingConfiguration();
            configuration.setVersion(5);
            MarshallerProvider provider=new DefaultMarshallerProvider(
                marshallerFactory, configuration);
            //MarshallingEncoder用于将实现序列化接口的POJO对象序列化为二进制字节数组
            MarshallingEncoder encoder=new MarshallingEncoder(provider);
            return encoder;
        }
    }

2、服务端代码：

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 SubReqServer
    {
        public void bind(int port)throws Exception{
            EventLoopGroup bossGroup=new NioEventLoopGroup();
            EventLoopGroup workerGroup=new NioEventLoopGroup();
            try
            {
                ServerBootstrap b=new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 100)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer<SocketChannel>()
                {
                    @Override
                    protected void initChannel(SocketChannel ch)
                        throws Exception
                    {
                        ch.pipeline().addLast(
                            MarshallingCodeCFactory.buildMarshallingDecoder());
                        ch.pipeline().addLast(
                            MarshallingCodeCFactory.buildMarshallingeEncoder());
                        ch.pipeline().addLast(new SubReqServerHandler());
                    }
                });
                ChannelFuture f=b.bind(port).sync();
                f.channel().closeFuture().sync();
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
            finally{
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
        
        public static void main(String[] args)throws Exception
        {
            int port =8888;
            if (args!=null&&args.length>0)
            {
                port=Integer.valueOf(args[0]);
            }
            new SubReqServer().bind(port);
        }
    }

3、服务端网络I/O事件类：

import com.exp.netty.chapter07.SubscribeReq;
    import com.exp.netty.chapter07.SubscribeResp;
    
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class SubReqServerHandler extends ChannelHandlerAdapter
    {
        @Override
        public void channelRead(ChannelHandlerContext ctx,Object msg)throws Exception{
            SubscribeReq req=(SubscribeReq)msg;
            if ("test".equalsIgnoreCase(req.getUserName()))
            {
                System.out.println("Service accept client subscribe req : ["+req.toString()+"]");
            }
            ctx.writeAndFlush(resp(req.getSubReqID()));
        }
        
        private SubscribeResp resp(int subReqID){
            SubscribeResp resp=new SubscribeResp();
            resp.setSubReqID(subReqID);
            resp.setRespCode(0);
            resp.setDesc("Netty book order succeed, 3 days later, sent to the designated address");
            return resp;
        }
        
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
            cause.printStackTrace();
            ctx.close();
        }
    }

4、客户端代码：

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 SubReqClient
    {
        public void connect(int port,String host)throws Exception{
            EventLoopGroup g=new NioEventLoopGroup();
            try
            {
                Bootstrap b=new Bootstrap();
                b.group(g)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>()
                {
    
                    @Override
                    protected void initChannel(SocketChannel ch)
                        throws Exception
                    {
                        ch.pipeline().addLast(
                            MarshallingCodeCFactory.buildMarshallingDecoder());
                        ch.pipeline().addLast(
                            MarshallingCodeCFactory.buildMarshallingeEncoder());
                        ch.pipeline().addLast(new SubReqClientHandler());
                    }
                });
                ChannelFuture f=b.connect(host, port).sync();
                f.channel().closeFuture().sync();
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
            finally{
                g.shutdownGracefully();
            }
        }
        
        public static void main(String[] args)throws Exception
        {
            int port =8888;
            if (args!=null&&args.length>0)
            {
                port=Integer.valueOf(args[0]);
            }
            new SubReqClient().connect(port, "127.0.0.1");
        }
    }

5、客户端网络I/O事件类：

import com.exp.netty.chapter07.SubscribeReq;
    
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class SubReqClientHandler extends ChannelHandlerAdapter
    {
        public SubReqClientHandler(){
            
        }
        
        @Override
        public void channelActive(ChannelHandlerContext ctx){
            for (int i = 0; i < 10; i++ )
            {
                ctx.write(subReq(i));
            }
            ctx.flush();
        }
        
        private SubscribeReq subReq(int i){
            SubscribeReq req=new SubscribeReq();
            req.setAddress("杭州市西湖风景区");
            req.setPhoneNumber("13888888888");
            req.setProductName("Netty 权威指南");
            req.setSubReqID(i);
            req.setUserName("test");
            return req;
        }
        
        @Override
        public void channelRead(ChannelHandlerContext ctx,Object msg){
            System.out.println("Receive server response : ["+msg+"]");
        }
        
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx)throws Exception{
            ctx.flush();
        }
        
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
            cause.printStackTrace();
            ctx.close();
        }
    }

![20190110145238819.png][]

这里客户端连续发送了10条消息都成功得到响应，说明模拟的TCP的粘包/拆包场景结果正确，因此Netty的Marshalling编解码器支持半包和粘包的处理，因此我们只需要将Marshalling的编码器和解码器加入ChannelPipeline中就可以实现对Marshalling序列化的支持，而且也方便了与JBoss内部模块的远程通信。

参考书籍《Netty权威指南》

[jboss-marshalling-1.3.0.CR9.jar]: http://download.jboss.org/jbossmarshalling/jboss-marshalling-1.3.0.CR9.jar
[jboss-marshalling-serial-1.3.0.CR9.jar]: http://download.jboss.org/jbossmarshalling/jboss-marshalling-serial-1.3.0.CR9.jar
[https_www.jboss.org_jbossmarshalling_downloads]: https://www.jboss.org/jbossmarshalling/downloads
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70]: /images/20220328/2a5a6ef0415e4019bf95fe4a8ec44502.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 1]: /images/20220328/82f1278670cb47919f7e913b75bb4dc6.png
[20190110145238819.png]: /images/20220328/51b7215a51ed467baab4466810b9ec52.png