編解碼－marshalling

JBoss的Marshalling序列化框架，它是JBoss內部使用的序列化框架，Netty提供了Marshalling編碼和解碼器，方便用戶在Netty中使用Marshalling。

JBoss Marshalling是一個Java對象序列化包，對JDK默認的序列化框架進行了優化，但又保持跟java.io.Serializable接口的兼容，同時增加了一些可調的參數和附加的特性，這些參數和特性可通過工廠類進行配置。

import lombok.Data;

import java.io.Serializable;

@Data
public class SubscribeReq implements Serializable {

    /**
     * 默認的序列號ID
     */
    private static final long serialVersionUID = 1L;

    private int subReqID;

    private String userName;

    private String productName;

    private String phoneNumber;

    private String address;

    @Override
    public String toString() {
        return "SubscribeReq [subReqID=" + subReqID + ", userName=" + userName
                + ", productName=" + productName + ", phoneNumber="
                + phoneNumber + ", address=" + address + "]";
    }
}

import lombok.Data;

import java.io.Serializable;

@Data
public class SubscribeResp implements Serializable {

   /**
   * 默認序列ID
   */
 private static final long serialVersionUID = 1L;

   private int subReqID;

   private int respCode;

   private String desc;

   @Override
   public String toString() {
   return "SubscribeResp [subReqID=" + subReqID + ", respCode=" + respCode
     + ", desc=" + desc + "]";
   }
 }

編解碼工廠類：

import io.netty.handler.codec.marshalling.*;
import org.jboss.marshalling.MarshallerFactory;
import org.jboss.marshalling.Marshalling;
import org.jboss.marshalling.MarshallingConfiguration;

public final class MarshallingCodeCFactory {

    /**
     * 創建Jboss Marshalling解碼器MarshallingDecoder
     */
    public static MarshallingDecoder buildMarshallingDecoder() {
        //首先通過Marshalling工具類的getProvidedMarshallerFactory靜態方法獲取MarshallerFactory實例
        //參數“serial”表示創建的是Java序列化工廠對象，它由jboss-marshalling-serial-1.3.0.CR9.jar提供。
        final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
        //創建了MarshallingConfiguration對象
        final MarshallingConfiguration configuration = new MarshallingConfiguration();
        //將它的版本號設置為5
        configuration.setVersion(5);
        //然后根據MarshallerFactory和MarshallingConfiguration創建UnmarshallerProvider實例
        UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);
        //最后通過構造函數創建Netty的MarshallingDecoder對象
        //它有兩個參數，分別是UnmarshallerProvider和單個消息序列化后的最大長度。
        MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024);
        return decoder;
    }

    /**
     * 創建Jboss Marshalling編碼器MarshallingEncoder
     */
    public static MarshallingEncoder buildMarshallingEncoder() {
        final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
        final MarshallingConfiguration configuration = new MarshallingConfiguration();
        configuration.setVersion(5);
        //創建MarshallerProvider對象，它用於創建Netty提供的MarshallingEncoder實例
        MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration);
        //MarshallingEncoder用於將實現序列化接口的POJO對象序列化為二進制數組。
        MarshallingEncoder encoder = new MarshallingEncoder(provider);
        return encoder;
    }
}

服務端代碼示例：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
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 {
        // 配置服務端的NIO線程組
        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() {
                        @Override
                        public void initChannel(Channel ch) {
                            //通過工廠類創建MarshallingEncoder解碼器，並添加到ChannelPipeline.
                            ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
                            //通過工廠類創建MarshallingEncoder編碼器，並添加到ChannelPipeline中。
                            ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
                            ch.pipeline().addLast(new SubReqServerHandler());
                        }
                    });

            // 綁定端口，同步等待成功
            ChannelFuture f = b.bind(port).sync();

            // 等待服務端監聽端口關閉
            f.channel().closeFuture().sync();
        } finally {
            // 優雅退出，釋放線程池資源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                // 采用默認值
            }
        }
        new SubReqServer().bind(port);
    }
}

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

@ChannelHandler.Sharable
public class SubReqServerHandler extends ChannelHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        //經過解碼器handler ObjectDecoder的解碼，
        //SubReqServerHandler接收到的請求消息已經被自動解碼為SubscribeReq對象，可以直接使用。
        SubscribeReq req = (SubscribeReq) msg;
        if ("Lilinfeng".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();// 發生異常，關閉鏈路
    }
}

客戶端代碼示例：

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

public class SubReqClient {

    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 ChannelInitializer() {
                        @Override
                        public void initChannel(Channel ch)
                                throws Exception {
                            ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
                            ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
                            ch.pipeline().addLast(new SubReqClientHandler());
                        }
                    });

            // 發起異步連接操作
            ChannelFuture f = b.connect(host, port).sync();

            // 等待客戶端鏈路關閉
            f.channel().closeFuture().sync();
        } finally {
            // 優雅退出，釋放NIO線程組
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                // 采用默認值
            }
        }
        new SubReqClient().connect(port, "127.0.0.1");
    }
}

import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

public class SubReqClientHandler extends ChannelHandlerAdapter {

    public SubReqClientHandler() {
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        //在鏈路激活的時候循環構造10條訂購請求消息，最后一次性地發送給服務端。
        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("138xxxxxxxxx");
        req.setProductName("Netty For Marshalling");
        req.setSubReqID(i);
        req.setUserName("Lilinfeng");
        return req;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        //由於對象解碼器已經對訂購請求應答消息進行了自動解碼，
        //因此，SubReqClientHandler接收到的消息已經是解碼成功后的訂購應答消息。
        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();
    }
}

運行結果：

服務端結果：

14:48:45.475 [nioEventLoopGroup-2-1] INFO i.n.handler.logging.LoggingHandler - [id: 0x71ea5def, /0:0:0:0:0:0:0:0:8080] RECEIVED: [id: 0x876eb7b4, /127.0.0.1:57423 => /127.0.0.1:8080]
14:48:45.707 [nioEventLoopGroup-3-1] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.leakDetectionLevel: simple
Service accept client subscribe req : [SubscribeReq [subReqID=0, userName=Lilinfeng, productName=Netty For Marshalling, phoneNumber=138xxxxxxxxx, address=南京市江寧區方山國家地質公園]]
Service accept client subscribe req : [SubscribeReq [subReqID=1, userName=Lilinfeng, productName=Netty For Marshalling, phoneNumber=138xxxxxxxxx, address=南京市江寧區方山國家地質公園]]
..........................................................................
Service accept client subscribe req : [SubscribeReq [subReqID=9, userName=Lilinfeng, productName=Netty For Marshalling, phoneNumber=138xxxxxxxxx, address=南京市江寧區方山國家地質公園]]

客戶端結果：

Receive server response : [SubscribeResp [subReqID=0, respCode=0, desc=Netty book order succeed, 3 days later, sent to the designated address]]
..........................................................................
Receive server response : [SubscribeResp [subReqID=9, respCode=0, desc=Netty book order succeed, 3 days later, sent to the designated address]]

由於我們模擬了TCP的粘包/拆包場景，但是程序的運行結果仍然正確，說明Netty的Marshalling編解碼器支持半包和粘包的處理，對於開發者而言，只需要正確地將Marshalling編碼器和解碼器加入到ChannelPipeline中，就能實現對Marshalling序列化的支持。

利用Netty的Marshalling編解碼器，可以輕松地開發出與JBoss內部模塊進行遠程通信的程序，而且支持異步非阻塞，這無疑降低了基於Netty開發的應用程序與JBoss內部模塊對接的難度。

pom.xml

        <dependency>
            <groupId>org.jboss.marshalling</groupId>
            <artifactId>jboss-marshalling</artifactId>
            <version>1.3.0.GA</version>
        </dependency>
        <dependency>
            <groupId>org.jboss.marshalling</groupId>
            <artifactId>jboss-marshalling-serial</artifactId>
            <version>1.3.0.GA</version>
        </dependency>