Netty之Marshalling编解码

左手的ㄟ右手 2023-06-19 13:27 8阅读 0赞

Netty主流的编解码器：

> ①JBoss的Marshalling包
> 
> ②google的Protobuf
> 
> ③基于Protobuf的Kyro
> 
> ④Apache的Thrift

本篇博客主要介绍Netty基于Marshalling的最简单的代码实践

实现引入依赖

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

MarshallingCodeFactory

package com.yj.service.codec;
    
    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;
    
    public class MarshallingCodeFactory {
    
    	/**
    	 * 创建Marshalling解码器MarshallingDecoder
    	 * 
    	 * @return
    	 */
    	public static MarshallingDecoder buildMarshallingDecoder() {
    		MarshallerFactory factory = Marshalling.getProvidedMarshallerFactory("serial");
    		MarshallingConfiguration configuration = new MarshallingConfiguration();
    		configuration.setVersion(5);
    		UnmarshallerProvider provider = new DefaultUnmarshallerProvider(factory, configuration);
    		MarshallingDecoder decoder = new MarshallingDecoder(provider);
    		return decoder;
    	}
    
    	/**
    	 * 创建MarshallingEncoder
    	 * 
    	 * @return
    	 */
    	public static MarshallingEncoder buildMarshallingEncoder() {
    		// 这里表示的是支持java serial对象的序列化。所以我们传输的对象要实现Serializable接口
    		MarshallerFactory factory = Marshalling.getProvidedMarshallerFactory("serial");
    		MarshallingConfiguration configuration = new MarshallingConfiguration();
    		configuration.setVersion(5);
    		MarshallerProvider provider = new DefaultMarshallerProvider(factory, configuration);
    		MarshallingEncoder encoder = new MarshallingEncoder(provider);
    		return encoder;
    
    	}
    }

客户端，服务端分别添加编解码器

pipeline.addLast(MarshallingCodeFactory.buildMarshallingEncoder());
    pipeline.addLast(MarshallingCodeFactory.buildMarshallingDecoder());

请求，响应实体都需要实现Serializable接口进行序列化

Req，Resp

package com.yj.entity;
    
    import java.io.Serializable;
    import lombok.Data;
    import lombok.ToString;
    
    @Data
    @ToString
    public class Req implements Serializable {
    	private static final long serialVersionUID = -1465519266979187295L;
    	private String id;
    	private String name;
    }

package com.yj.entity;
    
    import java.io.Serializable;
    import lombok.Data;
    import lombok.ToString;
    
    @Data
    @ToString
    public class Resp implements Serializable{
    	private static final long serialVersionUID = -1450604945238500417L;
    	private String id;
        private String name;
    }

客户端handler

package com.yj.service.codec;
    
    import com.yj.entity.Req;
    import com.yj.entity.Resp;
    
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    public class CodecClientHandler extends ChannelInboundHandlerAdapter {
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    		super.exceptionCaught(ctx, cause);
    	}
    
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    		Resp resp = (Resp) msg;
    		System.out.println("接收到服务端的响应:" + resp);
    	}
    
    	@Override
    	public void channelActive(ChannelHandlerContext ctx) throws Exception {
    		Req req = new Req();
    		req.setId("1");
    		req.setName("我的请求");
    		ctx.writeAndFlush(req).addListener(new ChannelFutureListener() {
    			public void operationComplete(ChannelFuture future) throws Exception {
    				if (future.isSuccess()) {
    					System.out.println("成功发送到服务端消息");
    				} else {
    					System.out.println("失败服务端消息失败:" + future.cause().getMessage());
    					future.cause().printStackTrace();
    				}
    			}
    		});
    	}
    }

服务端handler

package com.yj.service.codec;
    
    import com.yj.entity.Req;
    import com.yj.entity.Resp;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    public class CodecServerHandler extends ChannelInboundHandlerAdapter {
    
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg) {
    		Req req = (Req) msg;
    		System.out.println("接收到客户端的请求:" + req);
    	}
    
    	@Override
    	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    		Resp resp = new Resp();
    		resp.setId("1");
    		resp.setName("我的响应");
    		ctx.writeAndFlush(resp);
    	}
    }

分别启动服务端，客户端，观察输出结果

客户端输出：
    接收到服务端的响应:Resp(id=1, name=我的响应)
    服务端输出：
    接收到客户端的请求:Req(id=1, name=我的请求)

**服务端，客户端的Req和Resp实体似乎要位于相同的package下，否则解析会有异常**