netty编解码之jboss marshalling

jboss marshalling是jboss内部的一个序列化框架，速度也十分快，这里netty也提供了支持，使用十分方便，不需要像protobuf一样编写proto文件，只需要提供两个编解码器即可，以下就是jboss marshalling使用的开始。

源码程序

model类

model类和之前编写java序列化的时候没有区别，这里便不再多说，仅仅贴出源码：

SubscribeReq:

package cn.com.serialize;import java.io.Serializable;/*** Created by xiaxuan on 17/11/27.*/
public class SubscribeReq implements Serializable {private int subReqID;private String userName;private String productName;private String phoneNumber;private String address;//getter setter...@Overridepublic String toString() {return "SubscribeReq{" +"subReqID=" + subReqID +", userName='" + userName + '\'' +", productName='" + productName + '\'' +", phoneNumber='" + phoneNumber + '\'' +", address='" + address + '\'' +'}';}
}

SubscribeResp:

package cn.com.serialize;import java.io.Serializable;/*** Created by xiaxuan on 17/11/27.*/
public class SubscribeResp implements Serializable {private int subReqID;private int respCode;private String desc;//getter setter...@Overridepublic String toString() {return "SubscribeResp{" +"subReqID=" + subReqID +", respCode=" + respCode +", desc='" + desc + '\'' +'}';}
}

marshalling编解码工厂类

MarshallingCodeCFactory:

package cn.com.marshalling;import io.netty.handler.codec.marshalling.*;
import org.jboss.marshalling.MarshallerFactory;
import org.jboss.marshalling.Marshalling;
import org.jboss.marshalling.MarshallingConfiguration;/*** Created by xiaxuan on 17/11/28.*/
public final class MarshallingCodeCFactory {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, 1024);return decoder;}public static MarshallingEncoder buildMarshallingEncoder() {final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");final MarshallingConfiguration configuration = new MarshallingConfiguration();configuration.setVersion(5);MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration);MarshallingEncoder encoder = new MarshallingEncoder(provider);return encoder;}
}

这里在获取factory的时候传入的参数为serial表示为创建java序列化工厂对象。
在创建MarshallingDecoder对象时传入两个参数，分别是UnmarshallerProvider和单个消息序列化后的最大长度。

server程序

SubReqServer:

package cn.com.marshalling;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;/*** Created by xiaxuan on 17/11/28.*/
public class SubReqServer {public void bind(int port) {//配置服务端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<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());ch.pipeline().addLast(new SubReqServerHandler());}});//绑定端口，同步等待成功ChannelFuture f = b.bind(port).sync();//等待服务端监听端口关闭f.channel().closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();} finally {//优雅退出bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}public static void main(String[] args) {int port = 8080;new SubReqServer().bind(port);}
}

这里我们将先前MarshallingFactory工厂类创建的MarshallingDecoder解码器和MarshallingEncoder编码器加入到ChannelPipline中，来进行对我们的传输对象进行编解码。

SubReqServerHandler：

package cn.com.marshalling;import cn.com.serialize.SubscribeReq;
import cn.com.serialize.SubscribeResp;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;@Sharable
public class SubReqServerHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {SubscribeReq req = (SubscribeReq) msg;if ("Xiaxuan".equalsIgnoreCase(req.getUserName())) {System.out.println("Service accept client subscrib 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;}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();// 发生异常，关闭链路}
}

业务逻辑与之前并没有太多不同，均是获取到客户端请求后进行响应。

client程序

SubReqClient：

package cn.com.marshalling;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;/*** Created by xiaxuan on 17/12/5.*/
public class SubReqClient {public void connect(int port, String host) {EventLoopGroup group = new NioEventLoopGroup();try {Bootstrap b = new Bootstrap();b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel 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();} catch (InterruptedException e) {e.printStackTrace();} finally {group.shutdownGracefully();}}public static void main(String[] args) {int port = 8080;String host = "127.0.0.1";new SubReqClient().connect(port, host);}
}

除了编解码框架以外，其他的并没有不同的。

SubReqClientHandler:

package cn.com.marshalling;import cn.com.serialize.SubscribeReq;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;public class SubReqClientHandler extends ChannelInboundHandlerAdapter {public SubReqClientHandler() {}@Overridepublic 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("NanJing YuHuaTai");req.setPhoneNumber("135xxxxxxxx");req.setProductName("Netty Book For Marshalling");req.setSubReqID(i);req.setUserName("Xiaxuan");return req;}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {System.out.println("Receive server response : [" + msg + "]");}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();}
}

客户端业务逻辑与之前也没有其他的区别，都是构建十次请求，然后一次性发出，最后输出服务端的响应。

程序运行

分别启动server程序和client程序，查看运行结果，如下：

server:

client:

运行成功。

使用jboss marshalling来进行编解码还是比较简单的，在这里我们模拟了TCP的粘包/拆包场景，但是程序的运行结果仍然正确，说明Marshalling的编解码器支持半包和粘包的处理，对于普通的开发者来说，只需要将Marshalling编码器和解码器加入到ChannelPipline中，就能实现对Marshalling序列化的支持。