Netty实践与NIO原理

时间  2019-11-10
标签 netty 实践 nio 原理 栏目 Netty 繁體版
原文   原文链接

1、阻塞IO与非阻塞IOjava

Linux网络IO模型（5种）spring

（1）阻塞IO模型sql

全部文件操做都是阻塞的，以套接字接口为例，在进程空间中调用recvfrom，系统调用直到数据包到达且被复制到应用进程缓冲区或发生错误时才返回，期间会一直等待（阻塞）。模型如图：apache

（2）非阻塞IO模型bootstrap

recvfrom从应用层到内核时，若是该缓冲区没数据，直接返回一个EWOULDBLOCK错误，反复轮询检查这个状态，看是否有数据到来。如图：api

（3）IO复用模型数组

Linux提升select/poll，进程经过将一个或多个fd（file descriptor）传递给select或poll系统调用，阻塞在select操做上，侦测多个fd是否处于就绪状态。select/poll顺序扫描fd是否就绪，并且支持的fd数量有限。Linux还提供了一个epoll系统调用，使用基于事件驱动的方式代替顺序扫描，性能更高。当有fd就绪时，当即回调函数rollback。如图：安全

（4）信号驱动IO模型服务器

首先开启套接口信号驱动IO功能，经过系统调用sigaction执行一个信号处理函数，该函数当即返回，进程继续工做，它是非阻塞的。当数据准备就绪时，就为该进程生成一个SIGIO信号，经过信号回调通知应用程序调用recfrom来读取数据，通知主循环函数处理数据。如图：网络

（5）异步IO模型

告知内核启动某个操做，让内核在整个操做完成后（包括将数据从内核复制到用户本身的缓冲区）通知咱们。它与信号驱动的主要区别是：信号驱动IO由内核告知咱们什么时候开始一个IO操做，异步IO模型由内核通知咱们IO操做什么时候已经完成。如图所示：

IO多路复用的应用：

经过把多个IO的阻塞复用到一个select的阻塞上，使系统在单线程下可处理多个客户端请求。与传统多线程模型相比，最大优点是系统开销小，不须要建立额外进程或线程。主要应用场景以下：

（1）服务器须要同时处理多个处于监听状态或链接状态的套接字

（2）服务器须要同时处理多种网络协议的套接字

Linux最终选择epoll支持IO多路复用的系统调用，优势以下：

（1）支持一个进程打开的socket描述符（FD）不受限制（select单线程默认1024太少，epoll仅受限操做系统最大文件句柄数，1GB内存机器大约10万句柄）

（2）IO效率不会随FD数目增长而线性降低（只对“活跃”的socke进行t操做，活跃socket才会去主动调用callback函数）

（3）使用mmap加速内核与用户空间消息传递（同一块内存，避免没必要要复制）

（4）API简单：建立epoll描述符，添加监听事件，阻塞等待监听事件发生，关闭epoll描述符等

2、阻塞IO的例子（结合线程池）

//1.服务端

package com.xbai.io;

import java.io.IOException;

import java.net.ServerSocket;

import java.net.Socket;

import com.xbai.executor.TimeServerHandlerExecutePool;

import com.xbai.handler.TimeServerHandler;

public class TimeServerExecutor {

　　public static void main(String[] args)throws IOException {

　　　　int port =8080;

       if(args !=null && args.length >0){

　　　　　　try {

　　　　　　　　port = Integer.valueOf(args[0]);

       　　}catch (Exception e) {

　　　　　　　　// TODO: handle exception

       　　}

　　　　}

　　　　ServerSocket server =null;

      try {

　　　　　　server =new ServerSocket(port);

        　System.out.println("The time server is started in port : " + port);

        　TimeServerHandlerExecutePool singleExecutor =new TimeServerHandlerExecutePool(50,10000);

        　while(true){

　　　　　　　　Socket socket = server.accept();

            　singleExecutor.execute(new TimeServerHandler(socket));

        　}

　　　　}finally {

　　　　　　if(server !=null){

　　　　　　　　　System.out.println("The time server closed");

            　　server.close();

            　　server =null;

       　　}

　　　　}

　　}

}

//2.服务端线程池

package com.xbai.executor;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.ThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

public class TimeServerHandlerExecutePool {

　　private ExecutorService executor;

  　public TimeServerHandlerExecutePool(int maxPoolSize,int queueSize){

　　　　executor =new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),maxPoolSize,120L,TimeUnit.SECONDS,

              new ArrayBlockingQueue(queueSize));//线程池要执行的任务阻塞成一个队列，其内部的机制是等待唤醒生产者和消费者线程，有一个生产就可唤醒一个消费，去看源码的线程池原理

  　}

　　public void execute(Runnable task){

　　　　executor.execute(task);

  　}

}

//3.服务端处理器

package com.xbai.handler;

import java.io.BufferedReader;

import java.io.IOException;

import java.io.InputStreamReader;

import java.io.OutputStreamWriter;

import java.io.PrintWriter;

import java.net.Socket;

import java.sql.Date;

public class TimeServerHandler implements Runnable{

　　private Socketsocket;

  　public TimeServerHandler(Socket socket) {

　　　　this.socket = socket;

  　}

　　@Override

  　public void run() {

　　　　// TODO Auto-generated method stub

      BufferedReader br =null;

      PrintWriter pw =null;

      try {

　　　　　　br =new BufferedReader(new InputStreamReader(socket.getInputStream()));

        　pw =new PrintWriter(socket.getOutputStream(),true);

        　String curTime =null;

        　String msg =null;

        　while(true){

　　　　　　　　 msg = br.readLine();

            　if(msg ==null){

　　　　　　　　 break;

          　　}

　　　　　　System.out.println("The time server received order:" + msg);

          curTime ="query time order".equalsIgnoreCase(msg) ?new Date(

　　　　　　　　System.currentTimeMillis()).toString() :"bad order";

          pw.println(curTime);//这里不写println,就没法插入换行符，那边就不能readLine,一直阻塞，没法获取数据

        }

　　　}catch (IOException e) {

　　　　　if(br !=null){

　　　　　　　　try {

　　　　　　　　　　br.close();

          　　}catch (IOException e1) {

　　　　　　　　　　// TODO Auto-generated catch block

             　　e1.printStackTrace();

          　　}

　　　　　}

　　　　　if(pw !=null){

　　　　　　　　pw.close();

          　　pw =null;

     　　}

　　　　　if(socket !=null){

　　　　　　　　try {

　　　　　　　　　　socket.close();

          　　}catch (IOException e1) {

　　　　　　　　　　// TODO Auto-generated catch block

             　　e1.printStackTrace();

          　　}

　　　　　　　　socket =null;

     　　}

　　　}

　　}

}

//4.客户端代码

package com.xbai.io;

import java.io.BufferedReader;

import java.io.IOException;

import java.io.InputStreamReader;

import java.io.PrintWriter;

import java.net.Socket;

import java.net.UnknownHostException;

public class TimeClient {

　　public static void main(String[] args) {

　　　　int port =8080;

      if(args !=null && args.length >0){

　　　　　　try {

　　　　　　　　port = Integer.valueOf(args[0]);

        　}catch (Exception e) {

　　　　　　　　// TODO: handle exception

        　}

　　　　}

　　　　Socket socket =null;

      BufferedReader br =null;

      PrintWriter pw =null;

      try {

　　　　　socket =new Socket("localhost",port);

        br =new BufferedReader(new InputStreamReader(socket.getInputStream()));

        pw =new PrintWriter(socket.getOutputStream(),true);

        pw.println("query time order");

        System.out.println("send order succeed");

        String resp = br.readLine();

        System.out.println("Now is :" + resp);

      }catch (IOException e) {

　　　　　// TODO Auto-generated catch block

        e.printStackTrace();

      }finally{

　　　　　 if(pw !=null){

　　　　　 　　pw.close();

         　　pw =null;

      　　}

　　　　　 if(br !=null){

　　　　　　　try {

　　　　　　　　br.close();

       　　　}catch (IOException e) {

　　　　　　　　// TODO Auto-generated catch block

          　　e.printStackTrace();

       　　　}

　　　　  　　br =null;

     　　　}

　　　　　　if(socket !=null){

　　　　　　　　try {

　　　　　　　　　　socket.close();

       　　　　}catch (IOException e) {

　　　　　　　　　　// TODO Auto-generated catch block

          　　　　e.printStackTrace();

       　　　　}

　　　　　　　　socket =null;

     　　　}

　　　　}

　　}

}

执行结果

服务端启动及收发：

客户端发送和接收：

3、非阻塞IO的例子（原生Java NIO，目前有写半包等问题，怀疑服务端没有写出去致使的客户端Selector的关闭状态异常） 

//1.服务端主程序
package com.xiaobai.nio;

public class NIOServer {
    
    public static void main(String[] args) {
        MultiplexerTimeServer timeServer = new MultiplexerTimeServer();
        new Thread(timeServer,"NIO-MultiplexerTimeServer-001").start();
    }
}
//2.服务端timeServer
package com.xiaobai.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;

public class MultiplexerTimeServer implements Runnable {
    private Selector selector;
    private ServerSocketChannel servChannel;
    private volatile boolean stop;
    
    public MultiplexerTimeServer() {
        try {
            selector = Selector.open();//创建Selector
            servChannel = ServerSocketChannel.open();//创建Channel
            servChannel.configureBlocking(false);
            servChannel.socket().bind(new InetSocketAddress(2048), 1024);//ServerSocket绑定
            servChannel.register(selector, SelectionKey.OP_ACCEPT);//向Selector注册ACCEPT事件
            System.out.println("The time server is started in port 2048");
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    @Override
    public void run() {
        while(!stop){
            try {
                selector.select(1000);//轮询Channel
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.iterator();
                SelectionKey key = null;
                while(it.hasNext()){
                    key = it.next();
                    it.remove();//移除它
                    try {
                        handleInput(key);
                    } catch (Exception e) {
                        if(key != null){
                            key.cancel();
                            if(key.channel() != null){
                                key.channel().close();
                            }
                        }
                    }
                }
            } catch (IOException e1) {
                // TODO Auto-generated catch block
                e1.printStackTrace();
            }
        }
        if(selector != null){
            try {
                selector.close();
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    private void handleInput(SelectionKey key) throws IOException{
        if(key.isValid()){
            //处理新接入的请求
            if(key.isAcceptable()){//此前已向Selector注册，并已open
                //获取server channel
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                //获取client channel
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                //第一次捕捉到的客户端向Selector注册READ事件
                sc.register(selector, SelectionKey.OP_READ);
            }
            //处理已注册的读事件
            if(key.isReadable()){
                //获取客户端Channel
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);//读到缓冲
                if(readBytes > 0){
                    readBuffer.flip();
//                    Buffer java.nio.Buffer.flip()
//
//
//                    Flips this buffer. The limit is set to the current position and then the position is set to zero. If the mark is defined then it is discarded. 
//
//                    After a sequence of channel-read or put operations, invoke this method to prepare for a sequence of channel-write or relative get operations. For example: 
//
//                     buf.put(magic);    // Prepend header
//                     in.read(buf);      // Read data into rest of buffer
//                     buf.flip();        // Flip buffer
//                     out.write(buf);    // Write header + data to channel
                    byte[] bytes = new byte[readBuffer.remaining()];//缓冲中有多少个字节数据
                    readBuffer.get(bytes);
                    String body = new String(bytes,"UTF-8");
                    System.out.println("The time server receive order : " + body);
                    String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(//
                            System.currentTimeMillis()).toString() : "BAD ORDER";
                    doWrite(sc,currentTime);
                }else if(readBytes < 0){
                    //贵在坚持！
                    //对端链路关闭
//                    key.cancel();
//                    sc.close();
                }else{
                    ;//读到0字节，忽略
                }
            }
        }
    }
    private void doWrite(SocketChannel channel, String response) throws IOException{
        if(response != null && response.trim().length() > 0){
            byte[] bytes = response.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);//根据字节数组容量建立ByteBuffer
            writeBuffer.put(bytes);//字节数组复制到缓冲区
            writeBuffer.flip();
            channel.write(writeBuffer);//SocketChannel是异步非阻塞的，不保证一次发送完，出现“写半包”问题，
            //这里缺乏注册写操做，不断轮询Selector将没有发送完的ByteBuffer发送完毕
            //TODO 这里有问题，没有写出去，致使客户端没法收到消息，显示Selector关闭状态异常

        }
    }
}
//3.客户端主程序
package com.xiaobai.nio;

public class NIOClient {

    public static void main(String[] args) {
        TimeClientHandle timeClientHandle = new TimeClientHandle("127.0.0.1",2048);
        new Thread(timeClientHandle,"NIO-TimeClient-001").start();
    }
}
//4.客户端timeClient
package com.xiaobai.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;

public class TimeClientHandle implements Runnable {
    
    private String host;
    private int port;
    private Selector selector;
    private SocketChannel socketChannel;
    private volatile boolean stop;
    
    public TimeClientHandle(String host,int port) {
        this.host = host==null?"127.0.0.1":host;
        this.port = port;
        try {
            selector = Selector.open();
            socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);
        } catch (Exception e) {
            // TODO: handle exception
        }
        
    }
    @Override
    public void run() {
        try {
            doConnect();
        } catch (Exception e) {
            // TODO: handle exception
        }
        while(!stop){
            try {
                selector.select(3000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.iterator();
                SelectionKey key = null;
                while(it.hasNext()){
                    key = it.next();
                    it.remove();
                    try {
                        handleInput(key);
                    } catch (Exception e) {
                        if(key != null){
                            key.cancel();
                            if(key.channel() != null){
                                key.channel().close();
                            }
                        }
                    }
                }
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            if(selector != null){
                try {
                    selector.close();
                } catch (Exception e) {
                    // TODO: handle exception
                }
            }
        }
    }
    private void handleInput(SelectionKey key) throws Exception{
        if(key.isValid()){
            //判断是否链接成功
            //链接方法中已有链接不成功注册链接事件的逻辑，反复尝试链接，这里判断，若是成功，注册该客户链接的read事件准备接收数据
            SocketChannel sc = (SocketChannel) key.channel();
            if(key.isConnectable()){
                if(sc.finishConnect()){
                    sc.register(selector, SelectionKey.OP_READ);
                    doWrite(sc);//本客户向外写东西
                }
            }
            //下面是从服务器接收数据
            if(key.isReadable()){
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);//读到缓冲
                if(readBytes > 0){
                    readBuffer.flip();
//                    Buffer java.nio.Buffer.flip()
//
//
//                    Flips this buffer. The limit is set to the current position and then the position is set to zero. If the mark is defined then it is discarded.
//
//                    After a sequence of channel-read or put operations, invoke this method to prepare for a sequence of channel-write or relative get operations. For example:
//
//                     buf.put(magic);    // Prepend header
//                     in.read(buf);      // Read data into rest of buffer
//                     buf.flip();        // Flip buffer
//                     out.write(buf);    // Write header + data to channel
                    byte[] bytes = new byte[readBuffer.remaining()];//缓冲中有多少个字节数据
                    readBuffer.get(bytes);
                    String body = new String(bytes,"UTF-8");
                    System.out.println("Now is : " + body);
                    this.stop = true;
                }else if(readBytes < 0){
                    //贵在坚持！
                    //对端链路关闭
                    key.cancel();
                    sc.close();
                }else{
                    ;//读到0字节，忽略
                }
            }
        }
    }
    private void doConnect() throws IOException {
        //若是链接成功，则直接注册到多路复用器上，发送请求消息，读应答
        if(socketChannel.connect(new InetSocketAddress(host, port))){//异步链接，直至成功
            socketChannel.register(selector, SelectionKey.OP_READ);
            doWrite(socketChannel);
        }else{//注册链接事件，轮询直至链接成功
            //异步，究竟是什么概念？底层是什么原理？TCP/IP层面
            socketChannel.register(selector, SelectionKey.OP_CONNECT);
        }
    }
    private void doWrite(SocketChannel sc) throws IOException {
        //本客户向外写东西
        byte[] req = "QUERY TIME ORDER".getBytes();
        ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
        writeBuffer.put(req);
        writeBuffer.flip();
        sc.write(writeBuffer);
        if(!writeBuffer.hasRemaining()){
            System.out.println("Send order 2 server succeed.");
        }
    }

}

4、TCP与UDP

5、网络传输粘包与拆包问题

6、Netty入门案例与原理分析、Reactor模式

第一个例子：

//1.NettyServer
package com.xiaobai.server.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.net.InetSocketAddress;

public class NettyServer {

    private final int port;

    public NettyServer(int port) {
        this.port = port;
    }

    public static void main(String[] args) throws InterruptedException {

        if(args.length != 1) {
            System.err.println("Usage:" + NettyServer.class.getSimpleName() + " <port>");
            return;
        }
        int port = Integer.parseInt(args[0]);
        new NettyServer(port).start();
    }

    private void start() throws InterruptedException {
        final NettyServerHandler serverHandler = new NettyServerHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(group).channel(NioServerSocketChannel.class)
                    .localAddress(new InetSocketAddress(port))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(serverHandler);
                        }
                    });
            ChannelFuture f = b.bind().sync();
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully().sync();
        }
    }
}
View Code 
//2.NettyServerHandler
package com.xiaobai.server.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

@ChannelHandler.Sharable
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf in = (ByteBuf) msg;
        System.out.println("Server received:" + in.toString(CharsetUtil.UTF_8));
        ctx.write(in);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
                .addListener(ChannelFutureListener.CLOSE);//关闭该Channel
    }
}
View Code 
//3.NettyClient
package com.xiaobai.server.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;

public class NettyClient {

    private final String host;
    private final int port;

    public NettyClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public static void main(String[] args) throws InterruptedException {

        if(args.length != 2) {
            System.err.println("Usage:" + NettyServer.class.getSimpleName() + " <host> <port>");
            return;
        }
        String host = args[0];
        int port = Integer.parseInt(args[1]);
        new NettyClient(host,port).start();
    }

    private void start() throws InterruptedException {

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class)
                    .remoteAddress(new InetSocketAddress(host,port))
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new NettyClientHandler());
                        }
                    });
            ChannelFuture f = b.connect().sync();
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully().sync();
        }
    }

}
View Code 
//4.NettyClientHandler
package com.xiaobai.server.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelHandlerInvoker;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.EventExecutorGroup;

public class NettyClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

    @Override
    protected void messageReceived(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
        System.out.println("Client received: " + byteBuf.toString(CharsetUtil.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!",CharsetUtil.UTF_8));
    }
}
View Code

执行结果：

服务端：

客户端：

 
第二个例子： 
//服务端启动spring配置文件applicationContext.xml
<?xml version="1.0" encoding="UTF-8" ?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">

    <bean id="nettyServer" class="com.xiaobai.netty.server.NettyServer" init-method="start">
        <constructor-arg index="0" type="int">
            <value>8888</value>
        </constructor-arg>
    </bean>
</beans>
View Code 
//服务端启动类
package com.xiaobai.netty.spring;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class SpringStart {

    public static void main(String[] args) {

        ApplicationContext application = new ClassPathXmlApplicationContext("com/xiaobai/netty/spring/applicationContext.xml");
    }
}
View Code 
//NettyServer
package com.xiaobai.netty.server;

import com.xiaobai.netty.handlers.ChildChannelHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.apache.log4j.Logger;

public class NettyServer {

    private static final Logger logger = Logger.getLogger(NettyServer.class);

    //无参
    public NettyServer() {

    }
    //用于spring管理构造函数初始化bean
    public NettyServer(int port) {
        this.port = port;
    }

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private ServerBootstrap bootstrap;
    private int port;


    public void start() {
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
        try {
            bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,1024)
                    .childHandler(new ChildChannelHandler());
            //同步等待绑定端口成功
            ChannelFuture future = bootstrap.bind(port).sync();
            System.out.println("NettyServer Successfully Started in port:" + port);
            logger.info("NettyServer Successfully Started in port:" + port);
            //同步等待服务器端口关闭
            future.channel().closeFuture().sync();//经实验，这是阻塞方法，一直阻塞
        }catch (Exception e) {
            //优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

}
View Code 
//ChildChannelHandler
package com.xiaobai.netty.handlers;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

//注意参数化类型不要引错包
public class ChildChannelHandler extends ChannelInitializer<SocketChannel>{

    //原理来自平时读书实践（nio,aio,同步/异步原理，底层）
    //接收到客户端链接（可链接）后的初始化动做，使用责任链模式绑定一系列数据读写操做，用于可读可写时的操做
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
        socketChannel.pipeline().addLast(new StringDecoder());
        socketChannel.pipeline().addLast(new TimeServerHandler());//由简入难，不断调试、琢磨的框架
    }
}
View Code 
//TimeServerHandler
package com.xiaobai.netty.handlers;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

import java.util.Date;

public class TimeServerHandler extends ChannelHandlerAdapter {

    private int counter;
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//        ByteBuf buf = (ByteBuf) msg;
//        byte[] req = new byte[buf.readableBytes()];
//        buf.readBytes(req);
//        String body = new String(req,"UTF-8").substring(0,req.length - System.getProperty("line.separator").length());
        String body = (String)msg;
        System.out.println("The time server receive order:" + body + " ; the counter is : " + ++counter);
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() :"BAD ORDER";
        currentTime = currentTime + System.getProperty("line.separator");
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.write(resp);
    }

    //这些都是主程序代码中出现这些状况后调用的接口代码
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
}
View Code 
//客户端启动类
package com.xiaobai.netty.client;

import org.apache.log4j.Logger;

public class NettyTest {

    private static final Logger logger = Logger.getLogger(NettyTest.class);

    public static void main(String[] args) {
        try {
            logger.info("Netty communication start!");
            NettyClient client = new NettyClient("127.0.0.1",8888);
            client.send();
        }catch (Exception e) {
            logger.error("Client connected failure!");
        }

    }
}
View Code 
//NettyClient
package com.xiaobai.netty.client;

import com.xiaobai.netty.handlers.ChildChannelHandler;
import com.xiaobai.netty.handlers.TimeClientHandler;
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.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import org.apache.log4j.Logger;

public class NettyClient {

    private static final Logger logger = Logger.getLogger(NettyClient.class);
    //无参
    public NettyClient() {

    }
    //用于spring管理构造函数初始化bean
    public NettyClient(String host,int port) {
        this.host = host;
        this.port = port;
    }

    private EventLoopGroup group;
    private Bootstrap bootstrap;
    private int port;
    private String host;


    public void send() {
        try {
            group = new NioEventLoopGroup();
            if(host != null && host.trim() != "") {
                bootstrap = new Bootstrap();
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY,true)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
                                socketChannel.pipeline().addLast(new StringDecoder());
                                socketChannel.pipeline().addLast(new TimeClientHandler());
                            }
                        });
                //发起异步链接操做
                System.out.println("NettyClient connecting " + host +":" + port);
                ChannelFuture future = bootstrap.connect(host,port).sync();
                logger.info("NettyClient connected " + host +":" + port);
                //等待客户端关闭
                future.channel().closeFuture().sync();
            }else{
                logger.info("accessing nowhere!");
            }
        }catch (Exception e) {
            //优雅退出，释放线程池资源
            group.shutdownGracefully();
        }
    }

}
View Code 
//TimeClientHandler
package com.xiaobai.netty.handlers;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import org.apache.log4j.Logger;

public class TimeClientHandler extends ChannelHandlerAdapter {

    private static final Logger logger = Logger.getLogger(TimeClientHandler.class);

    private int counter;

    private byte[] req;

    public TimeClientHandler() {
        req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.warn("Unexpected exception from downstream:" + cause.getMessage());
        ctx.close();
    }

    //已链接后发送消息
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //不要反复用一个ByteBuf,经测试会出现没法发布的问题--上网多查，研究其中原理！！
        ByteBuf message = null;
        for(int i = 0;i < 100; i++) {
            message = Unpooled.buffer(req.length);
            message.writeBytes(req);
            ctx.writeAndFlush(message);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//        ByteBuf buf = (ByteBuf) msg;
//        byte[] req = new byte[buf.readableBytes()];
//        buf.readBytes(req);
//        String body = new String(req,"UTF-8");
        String body = (String)msg;
        System.out.println("Now is:" + body + " ; the counter is : " + ++counter);
    }
}
View Code

执行结果：

服务端：

 

客户端：

 

 

重要服务端组件类：

NioEventLoopGroup:一个线程组，包含了一组NIO线程，专门用于网络事件处理，实际上它们就是Reactor线程组。这里建立两个，一个用于服务端接受客户端的链接，另外一个用于SocketChannel的网络读写。

ServerBootstrap:Netty用于启动NIO服务端的辅助启动类，下降开发复杂度。它的group方法将两个NIO线程组看成入参传递到ServerBootstrap中。backlog:TCP参数，这里设置为1024.

NioServerSocketChannel:功能对应于JDK NIO类库中的ServerSocketChannel类。

ChildChannelHandler:绑定I/O事件的处理类，做用相似于Reactor模式中的Handler类，主要用于处理网络I/O事件，例如记录日志、对消息进行编解码等。

服务端启动辅助类配置完成后，调用它的bind方法绑定监听端口，随后调用它的同步阻塞方法sync等待绑定操做完成。完成以后Netty会返回一个ChannelFuture,它的功能相似于JDK的java.util.concurrent.Future,主要用于异步操做的通知回调。

future.channel().closeFuture.sync()是阻塞方法（一直阻塞，直到服务关闭），等待服务端链路关闭后才退出。

shutdownGracefully方法：优雅退出，释放关联资源。

ByteBuf:相似于JDK中的java.nio.ByteBuffer对象，不过它提供了更增强大和灵活的功能。经过ByteBuf的readableBytes方法可获取缓冲区可读字节数，根据可读字节数建立byte数组，经过ByteBuf的readBytes方法将缓冲区字节数组复制到新建byte数组中，经过ChannelHandlerContext的write方法异步发送应答消息给客户端。

ChannelHandlerContext的flush方法：将消息发送队列中的消息写入到SocketChannel中发送给对方。从性能角度考虑，为了防止频繁唤醒Selector进行消息发送（Writable事件），Netty的write方法并不直接将消息写入SocketChannel中，只是把待发送的消息放到发送缓冲数组中，再经过调用flush方法，将发送缓冲区中的消息所有写到SocketChannel中。

服务端建立、客户端接入源码与流程

1）建立ServerBootstrap实例。ServerBootstrap是Netty服务端启动辅助类，提供了一系列方法用于设置服务端启动相关参数，底层经过门面模式对各类能力进行抽象和封装，下降用户开发难度。ServerBootstrap只有一个无参构造函数，由于须要设置的参数太多了，且可能发生变化，故采用的是Builder模式。

2）设置并绑定Reactor线程池。ServerBootstrap的线程池是EventLoopGroup,实际就是EventLoop数组。EventLoop的职责是处理全部注册到本线程多路复用器Selector上的Channel,Selector轮询操做由绑定的EventLoop线程run方法驱动，在一个循环体内循环执行。EventLoop不只仅处理网络I/O事件，用户自定义Task和定时任务Task也统一由EventLoop负责处理，实现了线程模型统一。从调度层面看，也不存在从EventLoop线程中再启动其余类型线程用于异步执行其余任务。避免了多线程并发操做和锁竞争，提高了I/O线程的处理和调度性能。

建立两个EventLoopGroup（不是必需两个，可只建立一个共享），前一个是父，后一个是子。父被传入父类构造函数。

3）设置并绑定服务端Channel.做为NIO服务端，需建立ServerSocketChannel.Netty对原生NIO类库进行了封装，对应实现是NioServerSocketChannel.Netty的ServerBootstrap提供了channel方法用于指定服务端Channel类型，经过工厂类，利用反射建立NioServerSocketChannel对象。

指定NioServerSocketChannel后，设置TCP的backlog参数，底层C对应接口：

int listen(int fd,int backlog);

backlog指定内核为此套接口排队的最大链接个数。对于给定监听套接口，内核要维护两个队列：未连接队列和已链接队列，根据TCP三路握手过程当中的三个分节来分隔这两个队列。服务器处于listen状态时，收到客户端syn分节（connect）时在未完成队列中建立 一个新条目，而后后三路握手的第二个分节即服务器syn响应客户端，此条目在第三分节到达前（客户端对服务器syn的ack）一直保留在未完成 链接队列中。三路握手完成，条目从未完成链接队列搬到已完成链接队列尾部。当进程调用accept时，从已完成队列中的头部取出一个条目给进程。当已完成队列未空时进程将 睡眠，直到有条目才唤醒。backlog被规定为两个队列总和最大值。大多数实现默认为5，高并发下不够，未完成链接队列长度可能因客户端syn的到达及等待三路握手 第三分节的到达延时而增大。Netty默认backlog为100，用户可根据实际场景和网络情况 灵活设置。

4）链路创建时建立并初始化ChannelPipeline.ChannelPipeline不是NIO服务端必需，本质是一个负责处理网络事件的职责链，管理和执行ChannelHandler.网络事件以事件流形式流转，根据ChannelHandler执行策略调度执行。典型网络事件：

a.链路注册 b.链路激活 c.链路断开 d.接收到请求消息 e.请求消息接收并处理完毕 f.发送应答消息 g.链路发生异常 h.发生用户自定义事件

5）添加并设置ChannelHandler.ChannelHandler是Netty提供给用户定制和扩展的关键接口，例如消息编解码，心跳，安全认证，TSL/SSL认证，流量控制，流量整形等。Netty提供了大量系统ChannelHandler，比较实用的以下：

a.系统编解码框架：ByteToMessageCodec b.通用基于长度的半包解码器：LengthFieldBasedFrameDecoder c.码流日志打印：LoggingHandler d.SSL安全认证：SslHandler e.链路空闲检测：IdleStateHandler f.流量整形：ChannelTrafficShapingHandler g.Base64编解码：Base64Decoder和Base64Encoder

建立和添加ChannelHandler示例代码：

        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
        try {
            bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,1024)
                    .childHandler(new         ChannelInitializer<SocketChannel>(){
                        @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
        socketChannel.pipeline().addLast(new StringDecoder());
        socketChannel.pipeline().addLast(new TimeServerHandler());//由简入难，不断调试、琢磨的框架
                }
            });
            //同步等待绑定端口成功
            ChannelFuture future = bootstrap.bind(port).sync();
            System.out.println("NettyServer Successfully Started in port:" + port);
            logger.info("NettyServer Successfully Started in port:" + port);
            //同步等待服务器端口关闭
            future.channel().closeFuture().sync();//经实验，这是阻塞方法，一直阻塞
        }catch (Exception e) {
            //优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

用户可为启动辅助类和其父类分别指定Handler.两类Handler的用途不一样：子类Handler是NioServerSocketChannel对应的ChannelPipeline的Handler，父类中的Handler是客户端新接入的链接SocketChannel对应的ChannelPipeline的Handler.

本质区别就是：ServerBootstrap中的Handler是NioServerSocketChannel使用的，全部链接该监听端口的客户端都会执行它；父类AbstractBootstrap中的Handler是个工厂类，为每一个新接入的客户端都建立一个新的Handler.

6）绑定并启动监听端口。将ServerSocketChannel注册到Selector上监听客户端链接。

建立新的NioServerSocketChannel，两个参数：第一个参数是从父类NIO线程池中顺序获取的一个NioEventLoop，它就是服务端用于监听和接收客户端链接的Reactor线程，第二个参数是所谓workerGroup线程池，是处理I/O读写的Reactor线程组。

NioServerSocketChannel建立成功后的初始化：

a.设置Socket参数和NioServerSocketChannel附加属性 b.将AbstractBootstrap的Handler添加到NioServerSocketChannel的ChannelPipeline c.将用于服务端注册的Handler（ServerBootstrapAcceptor）添加到ChannelPipeline

NioServerSocketChannel注册：封装成Task投递到NioEventLoop，将NioServerSocketChannel注册到NioEventLoop的Selector上（此时注册0，不监放任何网络操做）

注册成功后，触发ChannelRegistered事件，判断监听是否成功，成功则触发ChannelActive事件，根据配置决定是否自动触发Channel的读事件，最终触发ChannelPipeline读操做，调用到HeadHandler的读方法（业务处理）。不一样Channel（不管客户端仍是服务端）对读操做准备工做不一样，所以doBeginRead是个多态方法，这里都要修改网络监听操做位为自身感兴趣的，NioServerSocketChannel感兴趣的为OP_ACCEPT（16）

用4 bit表示全部4种网络操做类型：OP_READ 0001 OP_WRITE 0010 OP_CONNECT 0100 OP_ACCEPT 1000

好处是方便经过位操做进行操做位判断和状态修改，提高操做性能。

7）Selector轮询。由Reactor线程NioEventLoop负责调度和执行Selector轮询操做，选择就绪的Channel集合。

根据就绪的操做位，执行不一样操做。NioServerSocketChannel监听的是链接操做，执行的是NioUnsafe（接口）的read方法，这里使用的是NioMessageUnsafe（实现类，还有一个NioByteUnsafe）

doReadMessages方法：实际就是接收新的客户端链接并建立NioSocketChannel.

接收到新的客户端链接后，触发ChannelPipeline的ChannelRead方法，执行headChannelHandlerContext的fireChannelRead方法（触发事件），事件在ChannelPipeline中传递，执行ServerBootstrapAcceptor的channelRead方法，该方法分三个步骤：

a.将启动时传入的childHandler加入到客户端SocketChannel的ChannelPipeline中 b.设置客户端SocketChannel的TCP参数 c.注册SocketChannel到多路复用器

NioSocketChannel注册：仍注册操做位为0，触发ChannelReadComplete事件，执行ChannelPipeline的read方法，执行HeadHandler的read方法，将网络操做位改成OP_READ.

到此，新接入客户端链接处理完成，可进行网络读写I/O操做

8）轮询到准备就绪的Channel后，由Reactor线程NioEventLoop执行ChannelPipeline相应方法（fire各类事件，触发各类ChannelHandler的事件回调，观察者模式），最终调度并执行ChannelHandler.

9）根据网络事件类型，调度执行Netty系统ChannelHandler和用户定制ChannelHandler.

7、Netty对粘包拆包的解决方案

 

8、编码与解码

9、序列化与反序列化

10、网络传输私有协议制定与聊天室业务实现

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程