java中channelmessage,MessagePack在Netty中的应用
[toc]
MessagePack在Netty中的應用
前面使用Netty通信時，傳輸的都是字符串對象，因為在進行遠程過程調用時，更多的是傳輸pojo對象，這時就需要對pojo對象進行序列化與反序列化(編碼與解碼)，因為Java序列化技術本身的局限性，所以往往會使用第三方的編解碼框架，如這里使用的MessagePack。
在使用MessagePack時，需要注意下面兩點：
MessagePack編碼后的結果是一個List對象；
傳輸的pojo對象一定要加上@Message注解，否則無法使用MessagePack進行編碼；
上面兩點確實非常重要，我第一次在Netty中使用MessagePack，因為沒有注意上面兩點，寫的Netty程序運行沒有報錯，客戶端連接服務端也沒有問題，但就是不能輸出傳輸的pojo對象，原因就是上面的這兩個問題，所以一定要先知道這兩點原理，否則后面在測試Netty程序時會有很多問題，并且排錯debug過程也不太容易。
下面就直接給出demo的代碼，因為在代碼中都加了很多注釋，所以這里不再詳細進行說明。
編碼器與×××
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)
* 它負責將Object類型的POJO對象編碼為byte數組，然后寫入到ByteBuf中
* @author yeyonghao
*
*/
public class MsgpackEncoder extends MessageToByteEncoder {
@Override
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
// 創建MessagePack對象
MessagePack msgpack = new MessagePack();
// 將對象編碼為MessagePack格式的字節數組
byte[] raw = msgpack.write(msg);
// 將字節數組寫入到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)
* 首先從數據報msg(數據類型取決于繼承MessageToMessageDecoder時填寫的泛型類型)中獲取需要解碼的byte數組
* 然后調用MessagePack的read方法將其反序列化(解碼)為Object對象
* 將解碼后的對象加入到解碼列表out中，這樣就完成了MessagePack的解碼操作
* @author yeyonghao
*
*/
public class MsgpackDecoder extends MessageToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List out) throws Exception {
// 從數據報msg中(這里的數據類型為ByteBuf，因為Netty的通信基于ByteBuf對象)
final byte[] array;
final int length = msg.readableBytes();
array = new byte[length];
/**
* 這里使用的是ByteBuf的getBytes方法來將ByteBuf對象轉換為字節數組，前面是使用readBytes，直接傳入一個接收的字節數組參數即可
* 這里的參數比較多，第一個參數是index，關于readerIndex，說明如下：
* ByteBuf是通過readerIndex跟writerIndex兩個位置指針來協助緩沖區的讀寫操作的，具體原理等到Netty源碼分析時再詳細學習一下
* 第二個參數是接收的字節數組
* 第三個參數是dstIndex the first index of the destination
* 第四個參數是length the number of bytes to transfer
*/
msg.getBytes(msg.readerIndex(), array, 0, length);
// 創建一個MessagePack對象
MessagePack msgpack = new MessagePack();
// 解碼并添加到解碼列表out中
out.add(msgpack.read(array));
}
}
服務端
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());
// 添加業務處理handler
ch.pipeline().addLast(new PojoServerHandler());
}
});
// 綁定端口，同步等待成功，該方法是同步阻塞的，綁定成功后返回一個ChannelFuture
ChannelFuture f = b.bind(port).sync();
// 等待服務端監聽端口關閉，阻塞，等待服務端鏈路關閉之后main函數才退出
f.channel().closeFuture().sync();
} finally {
// 優雅退出，釋放線程池資源
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類型，這點尤其需要注意
// 否則程序人執行，不會報錯，但沒有任何輸出
@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)
// 設置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());
// 添加業務處理handler
ch.pipeline().addLast(new PojoClientHandler());
}
});
// 發起異步連接操作(注意服務端是bind，客戶端則需要connect)
ChannelFuture f = b.connect(host, port).sync();
// 等待客戶端鏈路關閉
f.channel().closeFuture().sync();
} finally {
// 優雅退出，釋放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) {
// 采用默認值
}
}
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 + "]";
}
}
測試
運行服務端，再運行客戶端，服務端的輸出結果如下：
Pojo from client : ["client",10]
"client"
10
總結
以上是生活随笔為你收集整理的java中channelmessage,MessagePack在Netty中的应用的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 网页空间php可以赋值,js如何赋值给p
- 下一篇： java api集合,javaAPI_集