MessagePack介绍

目前主要的编解码：

1. Java序列化
2. Marshalig
3. XML
4. JSON
5. MessagePack
6. Protobuf
   …

本次编解码是使用 MessagePack实现的，MessagePack主要优点：

• 高效
• 跨语言
  官网解释：
  
  更多介绍请自行查看官网

MessagePack 使用

引入依赖

<dependency><groupId>org.msgpack</groupId><artifactId>msgpack</artifactId><version>0.6.12</version></dependency>

MessagePack 编解码使用起来非常方便，像这样

@org.junit.Testpublic void test() throws Exception{List<String> list = Lists.newArrayList();list.add("msgpack");list.add("kumofs");list.add("viver");MessagePack msgpack = new MessagePack();// Serializebyte[] raw = msgpack.write(list);// Deserialize directly using a template(直接使用模板反序列化)List<String> list1 = msgpack.read(raw, Templates.tList(Templates.TString));list.forEach(s -> System.out.println(s));}

基于 MessagePack 编解码器实现

编码器

public class MsgPackDecoder extends MessageToMessageDecoder<ByteBuf> {@Overrideprotected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {final byte[] array;final int length = byteBuf.readableBytes();array = new byte[length];// 获取数据包 byteBuf中的byte数组byteBuf.getBytes(byteBuf.readerIndex(), array, 0, length);MessagePack msgpack = new MessagePack();//调用 msgpack.read 反序列化然后加入到list中list.add(msgpack.read(array));}
}

解码器

/*** @author WH* @version 1.0* @date 2020/5/28 23:49* @Description MsgPackEncoder 继承 MessageToByteEncoder* 它负责将Object类型的POJO对象编码为byte数组，然后写入到ByteBuf中*/
public class MsgPackEncoder extends MessageToByteEncoder<Object> {@Overrideprotected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) throws Exception {MessagePack msgpack = new MessagePack();//Serializebyte[] raw = msgpack.write(o);byteBuf.writeBytes(raw);}}

客户端实现

EchoClient

public class EchoClient {public static void main(String[] args) {EventLoopGroup group = new NioEventLoopGroup();try {Bootstrap b = new Bootstrap();b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000).handler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(65535,0,2,0,2));ch.pipeline().addLast("msgpack decoder", new MsgPackDecoder());ch.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));ch.pipeline().addLast("msgpack encoder", new MsgPackEncoder());ch.pipeline().addLast(new EchoClientHandler(10));}});//发起异步连接操作ChannelFuture f = b.connect("127.0.0.1", 8080).sync();//等待客户端链路关闭f.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {//优雅退出，释放NIO线程租group.shutdownGracefully();}}
}

EchoClientHandler

@Slf4j
public class EchoClientHandler extends ChannelHandlerAdapter {private final int sendNumber;private static List<User> users;public EchoClientHandler(int sendNumber) {this.sendNumber = sendNumber;users = new ArrayList<>(sendNumber);for (int i = 0; i < sendNumber; i++) {String name = "阿离" + i;User user = new User(name, i);users.add(user);}}public void channelActive(ChannelHandlerContext ctx) throws Exception {for (User user : users) {ctx.write(user);log.info("客户端发送消息为" + user);ctx.writeAndFlush(user);}}public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.info("接收到服务器的消息: {}", msg);}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}}

服务端实现

EchoServer

public class EchoServer {public static void main(String[] args) {//配置服务端的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>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception{ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));ch.pipeline().addLast(new MsgPackDecoder());ch.pipeline().addLast(new LengthFieldPrepender(2));ch.pipeline().addLast(new MsgPackEncoder());ch.pipeline().addLast(new EchoServerHandler());}});//绑定端口，成功同步等待ChannelFuture f = b.bind(8080).sync();//订单服务端监听端口关闭f.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {//优雅退出，释放线程池资源bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

EchoServerHandler

@Slf4j
public class EchoServerHandler extends ChannelHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ctx.writeAndFlush(msg);log.info("服务端接受到的消息为：{}", msg);}
}

测试

源码下载

https://github.com/weihubeats/netty-student