Netty入门笔记

知识点：Reactor模式

一、Linux网络I/O模型简介

        1.阻塞I/O模型：开始I/O操作时直到数据包到达且被复制到应用进程的缓冲区中或者发生错误才返回。进程开始执行I/O操作到它返回的整段               时间内都是被阻塞的。
        2.非阻塞I/O模型：轮询检查是否有数据到来有的话就将数据报拷贝到用户空间（程序的空间中）并处理数据报
        3.I/O复用模型：linux提供一个select/poll，进程将多个fd传递给select或者poll给系统调用，select/poll帮我们检查是否有fd处于就绪状态。
        4.信号驱动I/O模型
        5.异步I/O

二、I/O多路复用技术 

三、NIO

        a.服务器端
            1.打开ServerSocketChannel
            2.绑定监听地址InetSocketAddress
            3.创建Selector,启动线程
            4.将ServerSocketChannel注册到Selector中并选择被关系的状态（Connect、Accept、read、write）
            5.如果有客户端接入则创建SocketChannel并将SocketChannel注册到Selector中并监听Read状态
            6.将SocketChannel中读取到的消息放入到Buffer中
            7.从Buffer中获取到数据并将响应数据放入到Buffer中，将Buffer中的数据通过管道传给客户端
            8.如果监听到接受连接状态时，则创建SocketChannel并设置SocketChannel的非阻塞模式，并将SocketChannel注册到Selector中并监听                 Read状态

public class Main { public static void main(String[] args) { int port = 8080; if (args != null && args.length > 0) { port = Integer.valueOf(args[0]); } MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port); new Thread(timeServer,"NIO-MultiplexerTimeServer-001").start(); }}

public class MultiplexerTimeServer implements Runnable { private Selector selector; private ServerSocketChannel servChannel; private volatile boolean stop; public MultiplexerTimeServer(int port) { try { selector = Selector.open(); servChannel = ServerSocketChannel.open(); servChannel.configureBlocking(false); servChannel.socket().bind(new InetSocketAddress(port), 1024); //将服务器绑定到selector中 并且选择器只会检查通道的是否接受到连接 servChannel.register(selector, SelectionKey.OP_ACCEPT); System.out.println("The time server is start in port : " + port); } catch (IOException e) { e.printStackTrace(); } } public void stop() { this.stop = true; } @Override public void run() { while (!stop) { try { //清空集合里的键重新依次询问已经注册的通道是否准备好选择器所感兴趣的某种操作，发现了则重新加入到集合中 selector.select(1000); Set<SelectionKey> selectionKeys = selector.selectedKeys(); Iterator<SelectionKey> it = selectionKeys.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) { e.printStackTrace(); } } if (selector != null) { try { selector.close(); } catch (IOException e) { e.printStackTrace(); } } } private void handleInput(SelectionKey key) throws IOException { if (key.isValid()) { //判断该键是否已经准备好接受连接 if (key.isAcceptable()) { ServerSocketChannel ssc = (ServerSocketChannel) key.channel(); SocketChannel sc = ssc.accept(); sc.configureBlocking(false); sc.register(selector, SelectionKey.OP_READ); } //判断该键是否已经准备好读取数据 if (key.isReadable()) { SocketChannel sc = (SocketChannel) key.channel(); ByteBuffer readBuffer = ByteBuffer.allocate(1024); int readBytes = sc.read(readBuffer); if (readBytes > 0) { readBuffer.flip(); 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 { ; } } } } private void doWrite(SocketChannel sc, String response) throws IOException { if (response != null && response.trim().length() > 0) { byte[] bytes = response.getBytes(); ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length); writeBuffer.put(bytes); writeBuffer.flip(); sc.write(writeBuffer); } }}

 b.客户端
            1.打开SocketChannel并设置非阻塞模式，同时设置TCP参数
            2.异步连接服务器
            3.如果还未连接成功则将SocketChannel注册到Selector中并监听Connect状态
            4.如果连接成功则将SocketChannel注册到Selector中并监听Read状态并发送数据给客户端
            5.当监听到连接状态时则判断是否已经完成连接如果未完成连接则发送数据给服务器
            6.如果连接已经完成则将数据发送给客户端并将SocketChannel注册进Selector并监听Read状态
            7.如果监听到读取状态则将服务器传过来的数据读取到buffer中再从Buffer将数据读取出来        

public class ClientMain { public static void main(String[] args) { int port = 8080; new Thread(new TimeClientHandle("127.0.0.1",port),"TimeClient-001").start(); }}

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; this.port = port; try { selector = Selector.open(); socketChannel = SocketChannel.open(); socketChannel.configureBlocking(false); } catch (IOException e) { e.printStackTrace(); } } @Override public void run() { try { doConnect(); } catch (IOException e) { e.printStackTrace(); } while (!stop) { try { selector.select(1000); 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) { e.printStackTrace(); } } if (selector != null) { try { selector.close(); } catch (IOException e) { e.printStackTrace(); } } } private void handleInput(SelectionKey key) throws IOException { if(key.isValid()){ SocketChannel sc = (SocketChannel) key.channel(); if(key.isConnectable()){ if(sc.finishConnect()){ sc.register(selector,SelectionKey.OP_READ); doWrite(sc); } else{ System.exit(1); } } if(key.isReadable()){ ByteBuffer readByteBuffer = ByteBuffer.allocate(1024); int readBytes = sc.read(readByteBuffer); if(readBytes>0){ readByteBuffer.flip(); byte[] bytes = new byte[readByteBuffer.remaining()]; readByteBuffer.get(bytes); String body = new String(bytes,"UTF-8"); System.out.println("Now is : "+body); this.stop = stop; } else if(readBytes<0){ key.cancel(); sc.close(); } else{ ; } } } } 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 succed."); } private void doConnect() throws IOException { if (socketChannel.connect(new InetSocketAddress(host, port))) { socketChannel.register(selector, SelectionKey.OP_READ); doWrite(socketChannel); } else { socketChannel.register(selector, SelectionKey.OP_CONNECT); } }}

四、Netty应用

五、TCP的粘包和拆包

六、Netty解决TCP的粘包和拆包的问题

        1.什么是粘包和拆包
            一个完整的数据包可能会由于数据量过大导致数据包被TCP拆成多个包发送，同时也有可能因为数据量过小而将多个包封装成一个大的数              据包发送，这就是所谓的TCP粘包和拆包问题
        2.TCP粘包和拆包发生的原因
            a.应用程序write写入的字节大小大于套接口发送缓冲区的大小；
            b.进行MSS大小的TCP分段；
            c.以太网帧的payload大于MTU进行IP分片；
        3.通过netty解决TCP粘包和拆包的问题
            a.使用LineBasedFrameDecoder解决TCP粘包和拆包问题
            b.使用DelimiterBasedFrameDecoder解决TCP粘包和拆包问题
            c.使用FiexedLengthFrameDecoder解决TCP粘包和拆包问题

七、编解码技术