Netty是建立在NIO基础之上，Netty在NIO之上又提供了更高层次的抽象。
在Netty里面，Accept连接可以使用单独的线程池去处理，读写操作又是另外的线程池来处理。
Accept连接和读写操作也可以使用同一个线程池来进行处理。而请求处理逻辑既可以使用单独的线程池进行处理，也可以跟放在读写线程一块处理。线程池中的每一个线程都是NIO线程。用户可以根据实际情况进行组装，构造出满足系统需求的并发模型。
Netty提供了内置的常用编解码器，包括行编解码器［一行一个请求］，前缀长度编解码器［前N个字节定义请求的字节长度］，可重放解码器［记录半包消息的状态］，HTTP编解码器，WebSocket消息编解码器等等
Netty提供了一些列生命周期回调接口，当一个完整的请求到达时，当一个连接关闭时，当一个连接建立时，用户都会收到回调事件，然后进行逻辑处理。
Netty可以同时管理多个端口，可以使用NIO客户端模型，这些对于RPC服务是很有必要的。
Netty除了可以处理TCP Socket之外，还可以处理UDP Socket。
在消息读写过程中，需要大量使用ByteBuffer，Netty对ByteBuffer在性能和使用的便捷性上都进行了优化和抽象。
代码：
服务端：
package com.kinson.netty.server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
/**
* descripiton:服务端
*
* @author: www.iknowba.cn
* @date: 2018/3/23
* @time: 15:37
* @modifier:
* @since:
*/
public class NettyServer {
/**
* 端口
*/
private int port;
public NettyServer(int port) {
this.port = port;
}
public void run() {
//EventLoopGroup是用来处理IO操作的多线程事件循环器
//负责接收客户端连接线程
EventLoopGroup bossGroup = new NioEventLoopGroup();
//负责处理客户端i/o事件、task任务、监听任务组
EventLoopGroup workerGroup = new NioEventLoopGroup();
//启动 NIO 服务的辅助启动类
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup);
//配置 Channel
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.childHandler(new ServerIniterHandler());
//BACKLOG用于构造服务端套接字ServerSocket对象，
// 标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
//是否启用心跳保活机制
bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
try {
//绑定服务端口监听
Channel channel = bootstrap.bind(port).sync().channel();
System.out.println("server run in port " + port);
//服务器关闭监听
/*channel.closeFuture().sync()实际是如何工作:
channel.closeFuture()不做任何操作，只是简单的返回channel对象中的closeFuture对象，对于每个Channel对象，都会有唯一的一个CloseFuture，用来表示关闭的Future，
所有执行channel.closeFuture().sync()就是执行的CloseFuturn的sync方法，从上面的解释可以知道，这步是会将当前线程阻塞在CloseFuture上*/
channel.closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
//关闭事件流组
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
new NettyServer(8899).run();
}
}
服务端业务逻辑处理：
package com.kinson.netty.server;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
/**
* descripiton: 服务器的处理逻辑
*
* @author: www.iknowba.cn
* @date: 2018/3/23
* @time: 15:50
* @modifier:
* @since:
*/
public class ServerHandler extends SimpleChannelInboundHandler<String> {
/**
* 所有的活动用户
*/
public static final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
/**
* 读取消息通道
*
* @param context
* @param s
* @throws Exception
*/
@Override
protected void channelRead0(ChannelHandlerContext context, String s)
throws Exception {
Channel channel = context.channel();
//当有用户发送消息的时候，对其他的用户发送消息
for (Channel ch : group) {
if (ch == channel) {
ch.writeAndFlush("[you]: " + s + "\n");
} else {
ch.writeAndFlush("[" + channel.remoteAddress() + "]: " + s + "\n");
}
}
System.out.println("[" + channel.remoteAddress() + "]: " + s + "\n");
}
/**
* 处理新加的消息通道
*
* @param ctx
* @throws Exception
*/
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
for (Channel ch : group) {
if (ch == channel) {
ch.writeAndFlush("[" + channel.remoteAddress() + "] coming");
}
}
group.add(channel);
}
/**
* 处理退出消息通道
*
* @param ctx
* @throws Exception
*/
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
for (Channel ch : group) {
if (ch == channel) {
ch.writeAndFlush("[" + channel.remoteAddress() + "] leaving");
}
}
group.remove(channel);
}
/**
* 在建立连接时发送消息
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
boolean active = channel.isActive();
if (active) {
System.out.println("[" + channel.remoteAddress() + "] is online");
} else {
System.out.println("[" + channel.remoteAddress() + "] is offline");
}
ctx.writeAndFlush("[server]: welcome");
}
/**
* 退出时发送消息
*
* @param ctx
* @throws Exception
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
if (!channel.isActive()) {
System.out.println("[" + channel.remoteAddress() + "] is offline");
} else {
System.out.println("[" + channel.remoteAddress() + "] is online");
}
}
/**
* 异常捕获
*
* @param ctx
* @param e
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
Channel channel = ctx.channel();
System.out.println("[" + channel.remoteAddress() + "] leave the room");
ctx.close().sync();
}
}
服务端处理器注册：
package com.kinson.netty.server;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
/**
* descripiton: 服务器初始化
*
* @author: www.iknowba.cn
* @date: 2018/3/23
* @time: 15:46
* @modifier:
* @since:
*/
public class ServerIniterHandler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//管道注册handler
ChannelPipeline pipeline = socketChannel.pipeline();
//编码通道处理
pipeline.addLast("decode", new StringDecoder());
//转码通道处理
pipeline.addLast("encode", new StringEncoder());
//聊天服务通道处理
pipeline.addLast("chat", new ServerHandler());
}
}
客户端：
package com.kinson.netty.client;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.commons.lang3.StringUtils;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
/**
* descripiton: 客户端
*
* @author: www.iknowba.cn
* @date: 2018/3/23
* @time: 16:40
* @modifier:
* @since:
*/
public class NettyClient {
private String ip;
private int port;
private boolean stop = false;
public NettyClient(String ip, int port) {
this.ip = ip;
this.port = port;
}
public void run() throws IOException {
//设置一个多线程循环器
EventLoopGroup workerGroup = new NioEventLoopGroup();
//启动附注类
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(workerGroup);
//指定所使用的NIO传输channel
bootstrap.channel(NioSocketChannel.class);
//指定客户端初始化处理
bootstrap.handler(new ClientIniterHandler());
try {
//连接服务
Channel channel = bootstrap.connect(ip, port).sync().channel();
while (true) {
//向服务端发送内容
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
String content = reader.readLine();
if (StringUtils.isNotEmpty(content)) {
if (StringUtils.equalsIgnoreCase(content, "q")) {
System.exit(1);
}
channel.writeAndFlush(content);
}
}
} catch (InterruptedException e) {
e.printStackTrace();
System.exit(1);
} finally {
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
new NettyClient("127.0.0.1", 8899).run();
}
}
客户端逻辑处理：
package com.kinson.netty.client;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
/**
* descripiton: 客户端逻辑处理
*
* @author: www.iknowba.cn
* @date: 2018/3/23
* @time: 16:50
* @modifier:
* @since:
*/
public class ClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
//打印服务端的发送数据
System.out.println(s);
}
}
客户端处理器注册：
package com.kinson.netty.client;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
/**
* descripiton: 客户端处理初始化
*
* @author: www.iknowba.cn
* @date: 2018/3/23
* @time: 16:55
* @modifier:
* @since:
*/
public class ClientIniterHandler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//注册管道
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
pipeline.addLast("http", new HttpClientCodec());
pipeline.addLast("chat", new ClientHandler());
}
}
测试时先启动服务端，再启动客户端。。。