在开始了解Netty是什么之前,我们先来回顾一下,如果我们需要实现一个客户端与服务端通信的程序,使用传统的IO编程,应该如何实现?
我们简化下场景:客户端每隔两秒发送一个带有时间戳的"hello world"给服务端,服务端收到之后打印.
public class IOServer {
public static void main(String[] args) throws Exception {
ServerSocket serverSocket = new ServerSocket(8000);
// (1) 接收新连接线程
new Thread(() -> {
while (true) {
try {
// (2) 阻塞方法获取新的连接
Socket socket = serverSocket.accept();
// (3) 每一个新的连接都创建一个线程，负责读取数据
new Thread(() -> {
try {
byte[] data = new byte[1024];
InputStream inputStream = socket.getInputStream();
while (true) {
int len;
// (3) 按字节流方式读取数据
while ((len = inputStream.read(data)) != -1) {
System.out.println(new String(data, 0, len));
}
}
} catch (IOException e) {
}
}).start();
} catch (IOException e) {
}
}
}).start();
}
}
server端首先创建了一个serverSocket来监听8000端口,然后创建一个线程,线程里面不断调用阻塞方法serverSocket.accept();获取新的连接,见(1),当获取到新的连接之后,给每条线程创建一个新的线程,这个线程负责从该连接中读取数据,见(2),然后读取数据是以字节流的方式,见(3).
public class IOClient {
public static void main(String[] args) {
new Thread(() ->{
try {
Socket socket = new Socket("127.0.0.1",8000);
while(true){
try{
socket.getOutputStream().write((new Date() + "hello world").getBytes());
socket.getOutputStream().flush();
Thread.sleep(2000);
}catch (Exception e){
}
}
} catch (IOException e) {
e.printStackTrace();
}
}).start();
}
}
客户端的代码相对简单,连接上服务端8000端口之后,每隔2秒,我们向服务端写一个带有时间戳的"hello world"
下面简单描述一下NIO是如何解决以上三个问题的.
NIO编程模型中,新来一个连接不在创建一个新的线程,而是可以把这条连接直接绑定到某个固定的线程,然后这条连接所有的读写都由这个线程来负责,那么他是怎么做到的?我们用一副图来对比一下IO与NIO
如上图所示,IO模型中,一个连接来了,会创建一个线程,对应一个while死循环,死循环的目的就是不断监测这条连接上是否有数据可读,大多数情况下,1W个连接里面同一时刻只有少量的连接有数据可读,因此,很多个while死循环就能监测1w个连接是否有数据可读的呢?
由于NIO模型中线程数量达达降低,线程切换效率因此也大幅度提高
NIO解决这个问题的方式是数据读写不在以字节为单位,而是以字节块为单位.IO模型中,每次都是以操作系统底层一个字节一个字节地读取数据,而NIO维护一个缓冲区,每次可以从这个缓冲区里面读取一块的数据,这就好比一盘美味的豆子放在那你的面前,你用筷子一个一个夹(每次一个),肯定不如要勺子挖着吃(每次一批)效率来的高
public class NioServer {
public static void main(String[] args) throws IOException {
Selector serverSelector = Selector.open();
Selector clientSelector = Selector.open();
new Thread(() ->{
try {
// 对应IO编程中的服务端启动
ServerSocketChannel listenerChannel = ServerSocketChannel.open();
listenerChannel.socket().bind(new InetSocketAddress(8000));
listenerChannel.configureBlocking(false);
listenerChannel.register(serverSelector, SelectionKey.OP_ACCEPT);
while(true){
// 监测是否有新的连接，这里的1指的是阻塞的时间为1ms
if(serverSelector.select(1) > 0){
Set<SelectionKey> set = serverSelector.selectedKeys();
Iterator<SelectionKey> keyIterator = set.iterator();
while(keyIterator.hasNext()){
SelectionKey key = keyIterator.next();
if(key.isAcceptable()){
try{
// (1)每来一个新连接，不需要创建一个线程,而是直接注册到clientSelector
SocketChannel clientChannel = (SocketChannel) key.channel();
clientChannel.configureBlocking(false);
clientChannel.register(clientSelector,SelectionKey.OP_ACCEPT);
}finally {
keyIterator.remove();
}
}
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}).start();
new Thread(() ->{
try{
while(true){
// (2) 批量轮询是否有哪些连接有数据可读，这里的1指的是阻塞的时间为1ms
if(clientSelector.select(1) > 0){
Set<SelectionKey> set = clientSelector.selectedKeys();
Iterator<SelectionKey> keyIterator = set.iterator();
while(keyIterator.hasNext()){
SelectionKey key = keyIterator.next();
if(key.isReadable()){
try{
SocketChannel clientChannel = (SocketChannel) key.channel();
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
// (3)读取数据以块为单位批量读取
clientChannel.read(byteBuffer);
byteBuffer.flip();
System.out.println(Charset.defaultCharset().newDecoder().decode(byteBuffer)
.toString());
}finally {
keyIterator.remove();
key.interestOps(SelectionKey.OP_READ);
}
}
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}).start();
}
}
相信大部分没有接触过NIO的同学应该会直接跳过代码来到这一行:原来使用JDK原生NIO的API实现一个简单的服务端通信程序是如此复杂.
那么Netty到底是何方神圣?
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.6.Final</version>
</dependency>
这么一小段代码就实现了我们前面NIO编程中的所有的功能,包括服务端启动,接收新连接,打印客户端传来的数据,.
public class TimeClient {
public static void main(String[] args) throws Exception {
String host = args[0];
int port = Integer.parseInt(args[1]);
EventLoopGroup workerGroup = new NioEventLoopGroup(); // (1)
try {
Bootstrap b = new Bootstrap(); // (2)
b.group(workerGroup); // (3)
b.channel(NioSocketChannel.class); // (4)
b.option(ChannelOption.SO_KEEPALIVE, true); // (5)
b.handler(new ChannelInitializer<SocketChannel>() { // (6)
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new TimeClientHandler());
}
});
// Start the client.
ChannelFuture f = b.connect(host, port).sync(); // (7)
// Wait until the connection is closed.
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
}
}
}
1.初始化用于连接即I/O工作的"线程池"
下面是Netty服务端实现部分:
EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap(); // (2)
b.group(bossGroup, workerGroup) // (3)
.channel(NioServerSocketChannel.class) // (4)
.handler(new LoggingHandler()) // (5)
.childHandler(new ChannelInitializer<SocketChannel>() { // (6)
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new DiscardServerHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128) // (7)
.childOption(ChannelOption.SO_KEEPALIVE, true); // (8)
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(port).sync(); // (9)
// Wait until the server socket is closed.
// In this example, this does not happen, but you can do that to gracefully
// shut down your server.
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
1.初始化用于Acceptor的主"线程池"以用于I/O工作的从"线程池"
转播
