Netty实战 第10章
1.jdk序列化
代码示例:
1.1 Server端
public class SubReqServer {
public void bind(int port) 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
public void initChannel(SocketChannel ch) {
ch.pipeline()
.addLast(
new ObjectDecoder(
1024 * 1024,
ClassResolvers
.weakCachingConcurrentResolver(this
.getClass()
.getClassLoader())));
ch.pipeline().addLast(new ObjectEncoder());
ch.pipeline().addLast(new SubReqServerHandler());
}
});
// 绑定端口，同步等待成功
ChannelFuture f = b.bind(port).sync();
// 等待服务端监听端口关闭
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(args[0]);
} catch (NumberFormatException e) {
// 采用默认值
}
}
new SubReqServer().bind(port);
}
}
public class SubReqServerHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
SubscribeReq req = (SubscribeReq) msg;
if ("Lilinfeng".equalsIgnoreCase(req.getUserName())) {
System.out.println("Service accept client subscrib req : ["
+ req.toString() + "]");
ctx.writeAndFlush(resp(req.getSubReqID()));
}
}
private SubscribeResp resp(int subReqID) {
SubscribeResp resp = new SubscribeResp();
resp.setSubReqID(subReqID);
resp.setRespCode(0);
resp.setDesc("Netty book order succeed, 3 days later, sent to the designated address");
return resp;
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();// 发生异常，关闭链路
}
}
1.2 Client端
public class SubReqClient {
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)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch)
throws Exception {
ch.pipeline().addLast(
new ObjectDecoder(1024, ClassResolvers
.cacheDisabled(this.getClass()
.getClassLoader())));
ch.pipeline().addLast(new ObjectEncoder());
ch.pipeline().addLast(new SubReqClientHandler());
}
});
// 发起异步连接操作
ChannelFuture f = b.connect(host, port).sync();
// 当代客户端链路关闭
f.channel().closeFuture().sync();
} finally {
// 优雅退出，释放NIO线程组
group.shutdownGracefully();
}
}
/**
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
int port = 8080;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
// 采用默认值
}
}
new SubReqClient().connect(port, "127.0.0.1");
}
}
public class SubReqClientHandler extends ChannelHandlerAdapter {
/**
* Creates a client-side handler.
*/
public SubReqClientHandler() {
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
for (int i = 0; i < 10; i++) {
ctx.write(subReq(i));
}
ctx.flush();
}
private SubscribeReq subReq(int i) {
SubscribeReq req = new SubscribeReq();
req.setAddress("南京市雨花台区软件大道101号华为基地");
req.setPhoneNumber("138xxxxxxxxx");
req.setProductName("Netty 最佳实践和原理分析");
req.setSubReqID(i);
req.setUserName("Lilinfeng");
return req;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
System.out.println("Receive server response : [" + msg + "]");
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
1.3 ClassResolver
根据className返回Class,ClassLoaderClassResolver是默认实现, CachingClassResolver在其基础上添加了cache功能, ClassResolvers是ClassResolver的辅助类用于创建ClassResolver
public interface ClassResolver {
Class<?> resolve(String className) throws ClassNotFoundException;
}
class ClassLoaderClassResolver implements ClassResolver {
private final ClassLoader classLoader;
ClassLoaderClassResolver(ClassLoader classLoader) {
this.classLoader = classLoader;
}
@Override
public Class<?> resolve(String className) throws ClassNotFoundException {
try {
return classLoader.loadClass(className);
} catch (ClassNotFoundException e) {
return Class.forName(className, false, classLoader);
}
}
}
class CachingClassResolver implements ClassResolver {
private final Map<String, Class<?>> classCache;
private final ClassResolver delegate;
CachingClassResolver(ClassResolver delegate, Map<String, Class<?>> classCache) {
this.delegate = delegate;
this.classCache = classCache;
}
@Override
public Class<?> resolve(String className) throws ClassNotFoundException {
// Query the cache first.
Class<?> clazz;
clazz = classCache.get(className);
if (clazz != null) {
return clazz;
}
// And then try to load.
clazz = delegate.resolve(className);
classCache.put(className, clazz);
return clazz;
}
}
public final class ClassResolvers {
/**
* cache disabled
* @param classLoader - specific classLoader to use, or null if you want to revert to default
* @return new instance of class resolver
*/
public static ClassResolver cacheDisabled(ClassLoader classLoader) {
return new ClassLoaderClassResolver(defaultClassLoader(classLoader));
}
/**
* non-agressive non-concurrent cache
* good for non-shared default cache
*
* @param classLoader - specific classLoader to use, or null if you want to revert to default
* @return new instance of class resolver
*/
public static ClassResolver weakCachingResolver(ClassLoader classLoader) {
return new CachingClassResolver(
new ClassLoaderClassResolver(defaultClassLoader(classLoader)),
new WeakReferenceMap<String, Class<?>>(new HashMap<String, Reference<Class<?>>>()));
}
}
2.StringDecoder
将ByteBuf对象转成字符串
public class StringDecoder extends MessageToMessageDecoder<ByteBuf> {
private final Charset charset;
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
out.add(msg.toString(charset));
}
}
解码后，返回的对象直接为String
3.MessageToMessageDecoder
StringDecoder继承自MessageToMessageDecoder,
MessageToMessageDecoder用于在消息格式之间进行转换,如下自定义Decoder，
将Integer(Integer在之前的ToIntegerDecoder做过转换)转为String
public class ToIntegerDecoder extends ByteToMessageDecoder {
@Override
public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
// 检查是否至少有4字节可读（一个int的字节长度）
if (in.readableBytes() >= 4) {
// 从入站ByteBuf 中读取一个int，并将其添加到解码消息的List 中
out.add(in.readInt());
}
}
}
public class IntegerToStringDecoder extends MessageToMessageDecoder<Integer> {
@Override
public void decode(ChannelHandlerContext ctx, Integer msg, List<Object> out) throws Exception {
// 将Integer 消息转换为它的String 表示，并将其添 加到输出的List 中
out.add(String.valueOf(msg));
}
}
public class ByteToCharServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ToIntegerDecoder());
ch.pipeline().addLast(new IntegerToStringDecoder());
ch.pipeline().addLast(new EchoServerHandler());
}
}
此时msg对象变成了String类型
