[toc]

MessagePack在Netty中的應(yīng)用

前面使用Netty通信時，傳輸?shù)亩际亲址畬ο?#xff0c;因?yàn)樵谶M(jìn)行遠(yuǎn)程過程調(diào)用時，更多的是傳輸pojo對象，這時就需要對pojo對象進(jìn)行序列化與反序列化(編碼與解碼)，因?yàn)镴ava序列化技術(shù)本身的局限性，所以往往會使用第三方的編解碼框架，如這里使用的MessagePack。

在使用MessagePack時，需要注意下面兩點(diǎn)：

MessagePack編碼后的結(jié)果是一個List對象；

傳輸?shù)膒ojo對象一定要加上@Message注解，否則無法使用MessagePack進(jìn)行編碼；

上面兩點(diǎn)確實(shí)非常重要，我第一次在Netty中使用MessagePack，因?yàn)闆]有注意上面兩點(diǎn)，寫的Netty程序運(yùn)行沒有報錯，客戶端連接服務(wù)端也沒有問題，但就是不能輸出傳輸?shù)膒ojo對象，原因就是上面的這兩個問題，所以一定要先知道這兩點(diǎn)原理，否則后面在測試Netty程序時會有很多問題，并且排錯debug過程也不太容易。

下面就直接給出demo的代碼，因?yàn)樵诖a中都加了很多注釋，所以這里不再詳細(xì)進(jìn)行說明。

編碼器與×××

MsgpackEncoder.java

package cn.xpleaf.msgpack;

import org.msgpack.MessagePack;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.handler.codec.MessageToByteEncoder;

/**

* MsgpackEncoder繼承自Netty中的MessageToByteEncoder類，

* 并重寫抽象方法encode(ChannelHandlerContext ctx, Object msg, ByteBuf out)

* 它負(fù)責(zé)將Object類型的POJO對象編碼為byte數(shù)組，然后寫入到ByteBuf中

* @author yeyonghao

*

*/

public class MsgpackEncoder extends MessageToByteEncoder {

@Override

protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {

// 創(chuàng)建MessagePack對象

MessagePack msgpack = new MessagePack();

// 將對象編碼為MessagePack格式的字節(jié)數(shù)組

byte[] raw = msgpack.write(msg);

// 將字節(jié)數(shù)組寫入到ByteBuf中

out.writeBytes(raw);

}

}

MsgpackDecoder

package cn.xpleaf.msgpack;

import java.util.List;

import org.msgpack.MessagePack;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.handler.codec.ByteToMessageDecoder;

import io.netty.handler.codec.MessageToMessageDecoder;

/**

* MsgpackDecoder繼承自Netty中的MessageToMessageDecoder類，

* 并重寫抽象方法decode(ChannelHandlerContext ctx, ByteBuf msg, List out)

* 首先從數(shù)據(jù)報msg(數(shù)據(jù)類型取決于繼承MessageToMessageDecoder時填寫的泛型類型)中獲取需要解碼的byte數(shù)組

* 然后調(diào)用MessagePack的read方法將其反序列化(解碼)為Object對象

* 將解碼后的對象加入到解碼列表out中，這樣就完成了MessagePack的解碼操作

* @author yeyonghao

*

*/

public class MsgpackDecoder extends MessageToMessageDecoder {

@Override

protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List out) throws Exception {

// 從數(shù)據(jù)報msg中(這里的數(shù)據(jù)類型為ByteBuf，因?yàn)镹etty的通信基于ByteBuf對象)

final byte[] array;

final int length = msg.readableBytes();

array = new byte[length];

/**

* 這里使用的是ByteBuf的getBytes方法來將ByteBuf對象轉(zhuǎn)換為字節(jié)數(shù)組，前面是使用readBytes，直接傳入一個接收的字節(jié)數(shù)組參數(shù)即可

* 這里的參數(shù)比較多，第一個參數(shù)是index，關(guān)于readerIndex，說明如下：

* ByteBuf是通過readerIndex跟writerIndex兩個位置指針來協(xié)助緩沖區(qū)的讀寫操作的，具體原理等到Netty源碼分析時再詳細(xì)學(xué)習(xí)一下

* 第二個參數(shù)是接收的字節(jié)數(shù)組

* 第三個參數(shù)是dstIndex the first index of the destination

* 第四個參數(shù)是length the number of bytes to transfer

*/

msg.getBytes(msg.readerIndex(), array, 0, length);

// 創(chuàng)建一個MessagePack對象

MessagePack msgpack = new MessagePack();

// 解碼并添加到解碼列表out中

out.add(msgpack.read(array));

}

}

服務(wù)端

PojoServer.java

package cn.xpleaf.basic;

import cn.xpleaf.msgpack.MsgpackDecoder;

import cn.xpleaf.msgpack.MsgpackEncoder;

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;

public class PojoServer {

public void bind(int port) throws Exception {

EventLoopGroup bossGroup = new NioEventLoopGroup();

EventLoopGroup workerGroup = new NioEventLoopGroup();

try {

ServerBootstrap b = new ServerBootstrap();

b.group(bossGroup, workerGroup)

.channel(NioServerSocketChannel.class)

.option(ChannelOption.SO_BACKLOG, 1024)

.childHandler(new ChannelInitializer() {

@Override

protected void initChannel(SocketChannel ch) throws Exception {

// 添加MesspagePack×××

ch.pipeline().addLast("msgpack decoder", new MsgpackDecoder());

// 添加MessagePack編碼器

ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder());

// 添加業(yè)務(wù)處理handler

ch.pipeline().addLast(new PojoServerHandler());

}

});

// 綁定端口，同步等待成功，該方法是同步阻塞的，綁定成功后返回一個ChannelFuture

ChannelFuture f = b.bind(port).sync();

// 等待服務(wù)端監(jiān)聽端口關(guān)閉，阻塞，等待服務(wù)端鏈路關(guān)閉之后main函數(shù)才退出

f.channel().closeFuture().sync();

} finally {

// 優(yōu)雅退出，釋放線程池資源

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(port);

} catch (NumberFormatException e) {

// TODO: handle exception

}

}

new PojoServer().bind(port);

}

}

PojoServerHandler.java

package cn.xpleaf.basic;

import java.util.List;

import cn.xpleaf.pojo.User;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

public class PojoServerHandler extends ChannelInboundHandlerAdapter {

@Override

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

// 注意msg為List，而不是User類型，這點(diǎn)尤其需要注意

// 否則程序人執(zhí)行，不會報錯，但沒有任何輸出

@SuppressWarnings("unchecked")

List list = (List) msg;

System.out.println("Pojo from client : " + list);

// 遍歷List，輸出的是pojo對象中的屬性

for (Object obj : list) {

System.out.println(obj);

}

}

@Override

public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

ctx.flush();

}

@Override

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

ctx.close();

}

}

客戶端

PojoClient.java

package cn.xpleaf.basic;

import cn.xpleaf.msgpack.MsgpackDecoder;

import cn.xpleaf.msgpack.MsgpackEncoder;

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;

public class PojoClient {

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)

// 設(shè)置TCP連接超時時間

.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)

.handler(new ChannelInitializer() {

@Override

protected void initChannel(SocketChannel ch) throws Exception {

// 添加MesspagePack×××

ch.pipeline().addLast("msgpack decoder", new MsgpackDecoder());

// 添加MessagePack編碼器

ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder());

// 添加業(yè)務(wù)處理handler

ch.pipeline().addLast(new PojoClientHandler());

}

});

// 發(fā)起異步連接操作(注意服務(wù)端是bind，客戶端則需要connect)

ChannelFuture f = b.connect(host, port).sync();

// 等待客戶端鏈路關(guān)閉

f.channel().closeFuture().sync();

} finally {

// 優(yōu)雅退出，釋放NIO線程組

group.shutdownGracefully();

}

}

public static void main(String[] args) throws Exception {

int port = 8080;

if(args != null && args.length > 0) {

try {

port = Integer.valueOf(port);

} catch (NumberFormatException e) {

// 采用默認(rèn)值

}

}

new PojoClient().connect(port, "localhost");

}

}

PojoClientHandler.java

package cn.xpleaf.basic;

import cn.xpleaf.pojo.User;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

public class PojoClientHandler extends ChannelInboundHandlerAdapter {

@Override

public void channelActive(ChannelHandlerContext ctx) {

User user = new User();

user.setName("client");

user.setAge(10);

// for(int i = 0; i < 10; i++) {

// ctx.write(user);

// }

// ctx.flush();

ctx.writeAndFlush(user);

}

@Override

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

}

@Override

public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

ctx.flush();

}

@Override

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

ctx.close();

}

}

POJO

User.java

package cn.xpleaf.pojo;

import org.msgpack.annotation.Message;

@Message

public class User {

private String name;

private int age;

public String getName() {

return name;

}

public void setName(String name) {

this.name = name;

}

public int getAge() {

return age;

}

public void setAge(int age) {

this.age = age;

}

@Override

public String toString() {

return "User [name=" + name + ", age=" + age + "]";

}

}

測試

運(yùn)行服務(wù)端，再運(yùn)行客戶端，服務(wù)端的輸出結(jié)果如下：

Pojo from client : ["client",10]

"client"

10

總結(jié)

以上是生活随笔為你收集整理的java中channelmessage,MessagePack在Netty中的应用的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。