Netty入门

/**
* 服务端处理通道.这里只是打印一下请求的内容，并不对请求进行任何的响应
* 继承自ChannelHandlerAdapter， 这个类实现了ChannelHandler接口，
* ChannelHandler提供了许多事件处理的接口方法，然后你可以覆盖这些方法。
* @author lcy
*
*/
public class DiscartServiceHandler extends ChannelHandlerAdapter {
/**
* 客户端收到新消息时，这个方法会被调用
*
* @param ctx
* 通道处理的上下文信息
* @param msg
* 接受的消息
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try {
// 将接收到的信息转换为缓冲区
ByteBuf str = (ByteBuf) msg;
// 打印传输过来的信息
System.out.print(str.toString(CharsetUtil.UTF_8));
} finally {
// 释放ByteBuf对象
ReferenceCountUtil.release(msg);
}
}/**
* 在异常时触发
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
//输出错误信息
cause.printStackTrace();
ctx.close();
}
}

/**
* 服务端
* @author lcy
*
*/
public class DiscartServer {
private int port;public DiscartServer(int port) {
super();
this.port = port;
}public void run() throws Exception {
//（一）设置两个线程组
//用来接收进来的连接
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
// 用来处理已经接受的连接
NioEventLoopGroup workGroup = new NioEventLoopGroup();
System.out.println("准备运行的端口" + port);
try {
//（二）辅助工具类，用于服务器通道的一系列配置
ServerBootstrap bootstrap = new ServerBootstrap();
//（三）绑定两个线程组
// 设置group，这一步是必须的，如果没有设置group将会报java.lang.IllegalStateException:group not set异常
bootstrap = bootstrap.group(bossGroup, workGroup);
//（四）指定NIO的模式
/***
* ServerSocketChannel以NIO的selector为基础进行实现的，用来接收新的连接
* 这里告诉Channel如何获取新的连接.
*/
bootstrap = bootstrap.channel(NioServerSocketChannel.class);
//（五）配置具体的数据处理方式,就是往里添加规则
bootstrap = bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Override
protected void initChannel(SocketChannel arg0) throws Exception {
//与50秒内都没有与服务端进行通信的客户端断连
arg0.pipeline().addLast(new ReadTimeoutHandler(50));
arg0.pipeline().addLast(new HttpObjectAggregator(1048576));
// 添加实际处理数据的类
arg0.pipeline().addLast(new DiscartServiceHandler());
}
});
//(六)设置TCP缓冲区
bootstrap = bootstrap.option(ChannelOption.SO_BACKLOG, 128);
//保持连接
bootstrap = bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
//（七）绑定端口，启动接收进来的连接
ChannelFuture sync = bootstrap.bind(port).sync();
//（八） 这里会一直等待，直到socket被关闭
sync.channel().closeFuture().sync();
} finally {
//（九）关闭资源
workGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
//服务开启
public static void main(String[] args) throws Exception {
int port;
if (args.length > 0) {
port = Integer.parseInt(args[0]);
} else {
port = 8080;
}
new DiscartServer(port).run();
System.out.println("server:run()");
}
}

public class ChannelClient extends ChannelInitializer{@Override
protected void initChannel(Channel arg0) throws Exception {
//与50秒内都没有与服务端进行通信的客户端断连
arg0.pipeline().addLast(new ReadTimeoutHandler(50));
arg0.pipeline().addLast(new HttpObjectAggregator(1048576));
//设置Channel
arg0.pipeline().addLast(new ChannelHandlerAdapter(){@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try {
// 将接收到的信息转换为缓冲区
ByteBuf str = (ByteBuf) msg;
// 打印传输过来的信息
System.out.print(str.toString(CharsetUtil.UTF_8));
} finally {
// 释放ByteBuf对象
ReferenceCountUtil.release(msg);
}
}@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
//输出错误信息
cause.printStackTrace();
ctx.close();
}
});}
}

/**
* 客户端
* @author lcy
*/
public class Client {
@SuppressWarnings("resource")
public static void main(String[] args) throws Exception {
//创建一个新的线程组
NioEventLoopGroup workGroup = new NioEventLoopGroup();
//初始化Netty
Bootstrap bootstrap = new Bootstrap();
//指定工作的线程组
bootstrap = bootstrap.group(workGroup);
//指定 Channel的类型。因为是客户端, 因此使用了 NioSocketChannel。
bootstrap.channel(NioSocketChannel.class);
/**
* 设置链接的一些属性
*/
//降低延迟，禁用了禁用nagle算法。nagle算法受TCP延迟确认影响,会导致相继两次向连接发送请求包。
bootstrap.option(ChannelOption.TCP_NODELAY, true);
//保持连接检测对方主机是否崩溃，避免（服务器）永远阻塞于TCP连接的输入
bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
//使用netty默认的解码器会出现读取不完整，不会执行channelRead方法。设置这个属性可惜保证Netty读取的完整
bootstrap.option(ChannelOption.MAX_MESSAGES_PER_READ, Integer.MAX_VALUE);
//设置数据处理器
bootstrap.handler(new ChannelClient());
//同步的链接
Channel channel = bootstrap.connect("127.0.0.1", 8080).sync().channel();
channel.writeAndFlush(Unpooled.copiedBuffer("Hello Netty...".getBytes()));
channel.closeFuture().sync();
workGroup.shutdownGracefully();
}
}