同步阻塞
式IO,服务端建立一个ServerSocket,而后客户端用一个Socket去链接那个ServerSocket,而后ServerSocket接收到一个Socket的链接请求就建立一个Socket和一个线程去跟那个Socket进行通讯。linux
public class BioServer { public static void main(String[] args) { // 服务端开启一个端口进行监听 int port = 8080; ServerSocket serverSocket = null; //服务端 Socket socket; //客户端 InputStream in = null; OutputStream out = null; try { serverSocket = new ServerSocket(port); //经过构造函数建立ServerSocket,指定监听端口,若是端口合法且空闲,服务器就会监听成功 // 经过无限循环监听客户端链接,若是没有客户端接入,则会阻塞在accept操做 while (true) { System.out.println("Waiting for a new Socket to establish" + " ," + new Date().toString()); socket = serverSocket.accept();//阻塞 三次握手 in = socket.getInputStream(); byte[] buffer = new byte[1024]; int length = 0; while ((length = in.read(buffer)) > 0) {//阻塞 System.out.println("input is:" + new String(buffer, 0, length) + " ," + new Date().toString()); out = socket.getOutputStream(); out.write("success".getBytes()); System.out.println("Server end" + " ," + new Date().toString()); } } } catch (Exception e) { e.printStackTrace(); } finally { // 必要的清理活动 if (serverSocket != null) { try { serverSocket.close(); } catch (IOException e) { e.printStackTrace(); } } if (in != null) { try { in.close(); } catch (IOException e) { e.printStackTrace(); } } if (out != null) { try { out.close(); } catch (IOException e) { e.printStackTrace(); } } } } }
同步非阻塞
包括Selector,这是多路复用器,selector会不断轮询注册的channel,若是某个channel上发生了读写事件,selector就会将这些channel获取出来,咱们经过SelectionKey获取有读写事件的channel,就能够进行IO操做。一个Selector就经过一个线程,就能够轮询成千上万的channel,这就意味着你的服务端能够接入成千上万的客户端。服务器
public class NioDemo implements Runnable { public int id = 100001; public int bufferSize = 2048; @Override public void run() { init(); } public void init() { try { // 建立通道和选择器 ServerSocketChannel socketChannel = ServerSocketChannel.open(); Selector selector = Selector.open(); InetSocketAddress inetSocketAddress = new InetSocketAddress( InetAddress.getLocalHost(), 4700); socketChannel.socket().bind(inetSocketAddress); // 设置通道非阻塞 绑定选择器 socketChannel.configureBlocking(false); socketChannel.register(selector, SelectionKey.OP_ACCEPT).attach( id++); System.out.println("Server started .... port:4700"); listener(selector); } catch (Exception e) { } } public void listener(Selector in_selector) { try { while (true) { Thread.sleep(1 * 1000); in_selector.select(); // 阻塞 直到有就绪事件为止 Set<SelectionKey> readySelectionKey = in_selector .selectedKeys(); Iterator<SelectionKey> it = readySelectionKey.iterator(); while (it.hasNext()) { SelectionKey selectionKey = it.next(); // 判断是哪一个事件 if (selectionKey.isAcceptable()) {// 客户请求链接 System.out.println(selectionKey.attachment() + " - 接受请求事件"); // 获取通道 接受链接, // 设置非阻塞模式(必须),同时须要注册 读写数据的事件,这样有消息触发时才能捕获 ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey .channel(); serverSocketChannel .accept() .configureBlocking(false) .register( in_selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE).attach(id++); System.out .println(selectionKey.attachment() + " - 已链接"); // 下面这种写法是有问题的 不该该在serverSocketChannel上面注册 /* * serverSocketChannel.configureBlocking(false); * serverSocketChannel.register(in_selector, * SelectionKey.OP_READ); * serverSocketChannel.register(in_selector, * SelectionKey.OP_WRITE); */ } if (selectionKey.isReadable()) {// 读数据 System.out.println(selectionKey.attachment() + " - 读数据事件"); SocketChannel clientChannel = (SocketChannel) selectionKey.channel(); ByteBuffer receiveBuf = ByteBuffer.allocate(bufferSize); clientChannel.read(receiveBuf); System.out.println(selectionKey.attachment() + " - 读取数据:" + getString(receiveBuf)); } if (selectionKey.isWritable()) {// 写数据 System.out.println(selectionKey.attachment() + " - 写数据事件"); SocketChannel clientChannel = (SocketChannel) selectionKey.channel(); ByteBuffer sendBuf = ByteBuffer.allocate(bufferSize); String sendText = "hello\n"; sendBuf.put(sendText.getBytes()); sendBuf.flip(); //写完数据后调用此方法 clientChannel.write(sendBuf); } if (selectionKey.isConnectable()) { System.out.println(selectionKey.attachment() + " - 链接事件"); } // 必须removed 不然会继续存在,下一次循环还会进来, // 注意removed 的位置,针对一个.next() remove一次 it.remove(); } } } catch (Exception e) { System.out.println("Error - " + e.getMessage()); e.printStackTrace(); } } /** * ByteBuffer 转换 String * * @param buffer * @return */ public static String getString(ByteBuffer buffer) { String string = ""; try { for (int i = 0; i < buffer.position(); i++) { string += (char) buffer.get(i); } return string; } catch (Exception ex) { ex.printStackTrace(); return ""; } } }
异步非阻塞
每一个链接发送过来的请求,都会绑定一个buffer,而后通知操做系统去异步完成读,此时你的程序是会去干别的事儿的,等操做系统完成数据读取以后,就会回调你的接口,给你操做系统异步读完的数据。网络
public class AIOServer { public final static int PORT = 9888; private AsynchronousServerSocketChannel server; public AIOServer() throws IOException { server = AsynchronousServerSocketChannel.open().bind( new InetSocketAddress(PORT)); } public void startWithFuture() throws InterruptedException, ExecutionException, TimeoutException { while (true) {// 循环接收客户端请求 Future<AsynchronousSocketChannel> future = server.accept(); AsynchronousSocketChannel socket = future.get();// get() 是为了确保 accept 到一个链接 handleWithFuture(socket); } } public void handleWithFuture(AsynchronousSocketChannel channel) throws InterruptedException, ExecutionException, TimeoutException { ByteBuffer readBuf = ByteBuffer.allocate(2); readBuf.clear(); while (true) {// 一次可能读不完 //get 是为了确保 read 完成,超时时间能够有效避免DOS攻击,若是客户端一直不发送数据,则进行超时处理 Integer integer = channel.read(readBuf).get(10, TimeUnit.SECONDS); System.out.println("read: " + integer); if (integer == -1) { break; } readBuf.flip(); System.out.println("received: " + Charset.forName("UTF-8").decode(readBuf)); readBuf.clear(); } } public void startWithCompletionHandler() throws InterruptedException, ExecutionException, TimeoutException { server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() { public void completed(AsynchronousSocketChannel result, Object attachment) { server.accept(null, this);// 再此接收客户端链接 handleWithCompletionHandler(result); } @Override public void failed(Throwable exc, Object attachment) { exc.printStackTrace(); } }); } public void handleWithCompletionHandler(final AsynchronousSocketChannel channel) { try { final ByteBuffer buffer = ByteBuffer.allocate(4); final long timeout = 10L; channel.read(buffer, timeout, TimeUnit.SECONDS, null, new CompletionHandler<Integer, Object>() { @Override public void completed(Integer result, Object attachment) { System.out.println("read:" + result); if (result == -1) { try { channel.close(); } catch (IOException e) { e.printStackTrace(); } return; } buffer.flip(); System.out.println("received message:" + Charset.forName("UTF-8").decode(buffer)); buffer.clear(); channel.read(buffer, timeout, TimeUnit.SECONDS, null, this); } @Override public void failed(Throwable exc, Object attachment) { exc.printStackTrace(); } }); } catch (Exception e) { e.printStackTrace(); } } public static void main(String args[]) throws Exception { // new AIOServer().startWithFuture(); new AIOServer().startWithCompletionHandler(); Thread.sleep(100000); } }
一种多路复用的技术,能够解决以前poll和select大量并发链接状况下cpu利用率太高,以及须要遍历整个被侦听的描述符集的问题。epoll只要遍历那些被内核IO事件异步唤醒而加入Ready队列的描述符集合就好了。并发
把一个磁盘文件映射到内存里来,而后把映射到内存里来的数据经过socket发送出去
有一种mmap技术,也就是内存映射,直接将磁盘文件数据映射到内核缓冲区,这个映射的过程是基于DMA引擎拷贝的,同时用户缓 冲区是跟内核缓冲区共享一块映射数据的,创建共享映射以后,就不须要从内核缓冲区拷贝到用户缓冲区了
光是这一点,就能够避免一次拷贝了,可是这个过程当中仍是会用户态切换到内核态去进行映射拷贝,接着再次从内核态切换到用户态, 创建用户缓冲区和内核缓冲区的映射
接着把数据经过Socket发送出去,仍是要再次切换到内核态
接着直接把内核缓冲区里的数据拷贝到Socket缓冲区里去,而后再拷贝到网络协议引擎里,发送出去就能够了,最后切换回用户态
减小一次拷贝,可是并不减小切换次数,一共是4次切换,3次拷贝异步
linux提供了sendfile,也就是零拷贝技术
这个零拷贝技术,就是先从用户态切换到内核态,在内核态的状态下,把磁盘上的数据拷贝到内核缓冲区,同时从内核缓冲区拷贝一些 offset和length到Socket缓冲区;接着从内核态切换到用户态,从内核缓冲区直接把数据拷贝到网络协议引擎里去
同时从Socket缓冲区里拷贝一些offset和length到网络协议引擎里去,可是这个offset和length的量不多,几乎能够忽略
只要2次切换,2次拷贝,就能够了socket
select
,poll
实现须要本身不断轮询全部fd集合,直到设备就绪,期间可能要睡眠和唤醒屡次交替。epoll
也须要调用epoll_wait不断轮询就绪链表,期间也可能屡次睡眠和唤醒交替,可是它是设备就绪时,调用回调函数,把就绪fd放入就绪链表中,并唤醒在epoll_wait中进入睡眠的进程。虽然都要睡眠和交替,可是select和poll在“醒着”的时候要遍历整个fd集合,而epoll在“醒着”的时候只要判断一下就绪链表是否为空就好了,这节省了大量的CPU时间。这就是回调机制带来的性能提高。ide
select
,poll
每次调用都要把fd集合从用户态往内核态拷贝一次,而且要把current往设备等待队列中挂一次,而epoll
只要一次拷贝,并且把current往等待队列上挂也只挂一次(在epoll_wait的开始,注意这里的等待队列并非设备等待队列,只是一个epoll内部定义的等待队列)。这也能节省很多的开销。函数