昨天晚上在看到7.2章MessagePack编码器和解码器开发这一章时，书里面没有贴出全部的代码，然后我按照我自己的想法把代码补全后，发现死活没有把代码跑通。
然后花了挺多时间在网上找，很多博客都贴出了这一节的代码，但是基本上都是把书上有的给贴出来了，严重怀疑他们敲完代码后有没有跑一遍。
不过最后还是找到了一个博客里面贴全了代码，发现是UserInfo类里面缺了一个注解@Message导致代码没跑通的。
下面贴上全部代码
UserInfo.java
package MessagePack; import org.msgpack.annotation.Message; @Message
public class UserInfo
{
private int age; private String name; public int getAge() {
return age;
} public void setAge(int age) {
this.age = age;
} public String getName() {
return name;
} public void setName(String name) {
this.name = name;
} @Override
public String toString()
{
return "age = " + age + "; name = " + name;
}
}
MsgpackDecoder.java
package MessagePack; import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;
import org.msgpack.MessagePack; import java.util.List; public class MsgpackDecoder extends MessageToMessageDecoder<ByteBuf>
{
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception
{
final byte[] array;
final int length = byteBuf.readableBytes();
array = new byte[length];
byteBuf.getBytes(byteBuf.readerIndex(), array, 0, length);
MessagePack msgpack = new MessagePack();
list.add(msgpack.read(array));
}
}
MsgpackEncoder.java
package MessagePack; import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.msgpack.MessagePack; public class MsgpackEncoder extends MessageToByteEncoder<Object>
{
protected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) throws Exception
{
MessagePack msgpack = new MessagePack();
byte[] raw = msgpack.write(o);
byteBuf.writeBytes(raw);
}
}
EchoServer.java
package MessagePack; 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 EchoServer
{
private final int port; public EchoServer(int port)
{
this.port = port;
} public void bind() throws Exception
{
//配置服务端的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>()
{
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception
{
socketChannel.pipeline().addLast("msgpack decoder", new MsgpackDecoder());
socketChannel.pipeline().addLast("msgpack encoder", new MsgpackEncoder());
socketChannel.pipeline().addLast(new EchoServerHandler());
}
});
//绑定端口，同步等待成功
ChannelFuture f = b.bind(port).sync(); //等待服务端监听端口关闭
System.out.println("bind");
f.channel().closeFuture().sync();
System.out.println("close");
}
finally
{
//优雅退出，释放线程池资源
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
} public static void main(String[] args) throws Exception
{
int port = 22233;
if (null != args && args.length > 0)
{
try
{
port = Integer.parseInt(args[0]);
}
catch (Exception e)
{
port = 22233;
}
} new EchoServer(port).bind();
}
}
EchoServerHandler.java
package MessagePack; import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext; public class EchoServerHandler extends ChannelHandlerAdapter
{
@Override
public void channelActive(ChannelHandlerContext ctx)
{
System.out.println("channelActive");
} @Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
{
System.out.println("Server receive the mspack message : " + msg);
ctx.writeAndFlush(msg);
} @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
{
System.out.println(cause.getMessage());
ctx.close();
}
}
EchoClient.java
package MessagePack; 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 EchoClient
{
private final String host;
private final int port;
private final int sendNumber; public EchoClient(String host, int port, int sendNumber)
{
this.host = host;
this.port = port;
this.sendNumber = sendNumber;
} public void connect() throws Exception
{
//配置客户端NIO线程组
EventLoopGroup group = new NioEventLoopGroup();
try
{
Bootstrap b = new Bootstrap();
b.group(group).channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>()
{
protected void initChannel(SocketChannel socketChannel) throws Exception
{
socketChannel.pipeline().addLast("msgpack decoder", new MsgpackDecoder());
socketChannel.pipeline().addLast("msgpack encoder", new MsgpackEncoder());
socketChannel.pipeline().addLast(new EchoClientHandler(sendNumber));
}
});
//发起异步连接操作
ChannelFuture f = b.connect(host, port).sync(); System.out.println("connect");
//等待客户端链路关闭
f.channel().closeFuture().sync();
System.out.println("close");
}
finally
{
//优雅退出，释放NIO线程组
group.shutdownGracefully();
}
} public static void main(String[] args) throws Exception
{
int port = 22233;
if (null != args && args.length > 0)
{
try
{
port = Integer.parseInt(args[0]);
}
catch (Exception e)
{
port = 22233;
}
} new EchoClient("127.0.0.1", 22233, 100).connect();
}
}
EchoClientHandler.java
package MessagePack; import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext; public class EchoClientHandler extends ChannelHandlerAdapter
{
private final int sendNumber; public EchoClientHandler(int sendNumber)
{
this.sendNumber = sendNumber;
} @Override
public void channelActive(ChannelHandlerContext ctx)
{
System.out.println("channelActive");
// ByteBuf byteBuf = Unpooled.copiedBuffer("asdf".getBytes());
// ctx.writeAndFlush(byteBuf); UserInfo[] infos = UserInfo();
for (UserInfo infoE : infos)
{
ctx.write(infoE);
}
ctx.flush();
} private UserInfo[] UserInfo()
{
UserInfo[] userInfos = new UserInfo[sendNumber];
UserInfo userInfo = null;
for (int i=0; i < sendNumber; i++)
{
userInfo = new UserInfo();
userInfo.setAge(i);
userInfo.setName("ABCDEFG --->" + i);
userInfos[i] = userInfo;
}
return userInfos;
} @Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
{
System.out.println("Client receive the msgpack message : " + msg);
//ctx.write(msg);
} @Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception
{
ctx.flush();
}
}