JBoss Marshalling也是在Netty框架中被经常使用的编解码技术，它对JDK默认的序列号框架做了优化，又保持和java.io.Serializable接口的兼容，有必要学习一下，现在看一下JBoss Marshalling在netty中是如何编解码的，对于半包的问题是否有处理呢？
示例编码环境：
springboot2.1.4
jdk1.8
首先，在pom中引入必要的架包：
<!-- jboss-marshalling编解码和序列号架包 -->
<dependency>
<groupId>org.jboss.marshalling</groupId>
<artifactId>jboss-marshalling</artifactId>
<version>2.0.9.Final</version>
</dependency>
<dependency>
<groupId>org.jboss.marshalling</groupId>
<artifactId>jboss-marshalling-serial</artifactId>
<version>2.0.9.Final</version>
</dependency>
第二步，服务端编码实现；
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;
/**
* JBoss Marshalling编解码服务端测试示例
* @author 程就人生
* @date 2019年10月26日
*/
public class MarshallingServer {
//创建线程组
private final EventLoopGroup bossGroup = new NioEventLoopGroup();
private final EventLoopGroup workerGroup = new NioEventLoopGroup();
/**
* 启动服务
* @param port
*/
public void start(int port){
try {
ServerBootstrap serverBootstrap = new ServerBootstrap().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 {
//JBoss解码
ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
//JBoss编码
ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
ch.pipeline().addLast(new MarshallingServerHandler());
}
});
//绑定端口，同步等等成功
ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
//等待服务端监听端口关闭
channelFuture.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally{
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
MarshallingServer server = new MarshallingServer();
server.start(7788);
}
}
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
* I/O消息的接收和处理
* @author 程就人生
* @date 2019年10月26日
*/
public class MarshallingServerHandler extends ChannelInboundHandlerAdapter{
private static Logger log = LoggerFactory.getLogger(MarshallingServerHandler.class);
/**
* 接收到消息时的处理
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Test test = (Test) msg;
if(test.getBody().contains("程就人生")){
log.info("接收到的消息：" + test.toString());
ctx.writeAndFlush(test);
}
}
/**
* 异常时关闭
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
//发生异常，关闭连接
ctx.close();
}
}
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;
/**
* JBoss 编解码实现
* @author 程就人生
* @date 2019年10月26日
*/
public final class MarshallingCodeCFactory {
/**
* 创建JBoss Marshalling解码器
* @return
*
*/
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*1024*1);
return decoder;
}
/**
* 创建JBoss Marshalling编码器
* @return
*
*/
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;
}
}
import java.io.Serializable;
/**
* java POJO 必须序列化
* @author 程就人生
* @date 2019年10月26日
*/
public class Test implements Serializable {
private String uid;
private String body;
public String getUid() {
return uid;
}
public void setUid(String uid) {
this.uid = uid;
}
public String getBody() {
return body;
}
public void setBody(String body) {
this.body = body;
}
@Override
public String toString() {
return "Test [uid=" + uid + ", body=" + body + "]";
}
}
第三步，客户端编码实现；
import com.example.demo.server5.MarshallingCodeCFactory;
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;
/**
* JBoss Marshalling编解码，客户端模拟
* @author 程就人生
* @date 2019年10月13日
*/
public class MarshallingClient {
public void connect(int port, String host){
EventLoopGroup group = new NioEventLoopGroup();
try{
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group).channel(NioSocketChannel.class)//设置为非阻塞
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>(){
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//JBoss解码
ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
//JBoss编码
ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
ch.pipeline().addLast(new MarshallingClientHandler());
}
});
//建立连接
ChannelFuture channelFuture = bootstrap.connect(host, port);
//等待服务端监听端口关闭
channelFuture.channel().closeFuture().sync();
}catch(Exception e){
e.printStackTrace();
}finally{
group.shutdownGracefully();
}
}
public static void main(String[] argo){
new MarshallingClient().connect(7788, "localhost");
}
}
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.example.demo.server5.Test;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
* I/O消息的接收和处理
* @author 程就人生
* @date 2019年10月26日
*/
public class MarshallingClientHandler extends ChannelInboundHandlerAdapter{
private static Logger log = LoggerFactory.getLogger(MarshallingClientHandler.class);
/**
*
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Test test = null;
for(int i=0;i<100;i++){
test = new Test();
test.setUid(i+"");
test.setBody("程就人生"+i);
ctx.writeAndFlush(test);
}
}
/**
* 接收到消息时的处理
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Test test = (Test) msg;
log.info("接收到的消息：" + test.toString());
}
/**
* 异常时关闭
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
ctx.close();
}
}
第四步，测试；先启动服务端，在启动客户端，为了了解Marshalling对半包的处理，代码里第一次连接服务器成功后，就向服务端主动发送一百次消息，看看服务端是否能够完整接收，完整返回；
服务端测试结果
客户端测试结果
从控制台的结果来看，服务器端很完整地接收到了客户端发送的消息，并没有发送粘包/半包的情况，客户端也完整地接收到了服务端发生的消息；通过本次测试，可以发现JBoss的Marshalling是支持半包处理的，在服务器端的hanlder和客户端的hanlder中只是加了编解码，并未加入其它的对半包处理的代码，使用起来还是很方便的。
网友评论