devbox
喵喵爱编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Netty网络通信之Socket_netty socket

作者：喵喵爱编程 | 2024-08-13 08:13:29

赞

踩

netty socket

一.什么是Socket

1.Socket起源于Unix，而Unix/Linux基本思想之一就是“一切皆文件”，也称为文件描述符

2.既然一切都是文件，那么就可以把对Socket的操作就是对“open—write/read—close”模式的一种实现

3.Socket是对TCP/IP协议的封装，Socket本身不是协议，通过Socket才能使用TCP/IP协议

二.Java四种IO模型

1.BIO（阻塞IO） 

  1. package com.lx.netty.bio;
  2.  
  3. import java.io.*;
  4. import java.net.ServerSocket;
  5. import java.net.Socket;
  6. import java.util.concurrent.ExecutorService;
  7. import java.util.concurrent.Executors;
  8.  
  9. public class BlockingServer {
  10.  
  11.     final static ExecutorService exec = Executors.newCachedThreadPool();
  12.  
  13.     public static void main(String[] args) {
  14.  
  15.         try {
  16.             // 监听端口
  17.             ServerSocket serverSocket = new ServerSocket(8080);
  18.             // 等待客户端的连接过来,如果没有连接过来，就会阻塞
  19.             while (true){
  20.                 // 阻塞IO中一个线程只能处理一个连接
  21.                 Socket socket = serverSocket.accept();
  22.                 exec.execute(() ->{
  23.                     // 获取数据
  24.                     String line = null;
  25.                     try {
  26.                         // 获取Socket中的输入流
  27.                         BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
  28.                         line = bufferedReader.readLine();
  29.                         System.out.println("客户端的数据："+ line);
  30.  
  31.                         BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
  32.                         bufferedWriter.write("ok\n");
  33.                         bufferedWriter.flush();
  34.                     } catch (IOException e) {
  35.                         e.printStackTrace();
  36.                     }
  37.  
  38.                 });
  39.  
  40.             }
  41.  
  42.  
  43.         } catch (IOException e) {
  44.             e.printStackTrace();
  45.         }
  46.     }
  47. }
  1. package com.lx.netty.bio;
  2.  
  3. import java.io.*;
  4. import java.net.Socket;
  5.  
  6. public class BlockingClient {
  7.  
  8.     public static void main(String[] args) {
  9.  
  10.         try {
  11.             // 建立连接
  12.             Socket socket = new Socket("localhost",8080);
  13.             // 向服务端写数据
  14.             BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
  15.             bufferedWriter.write("我是客户端，收到请回答！！\n");
  16.             bufferedWriter.flush();
  17.  
  18.             BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
  19.             String line = bufferedReader.readLine();
  20.             System.out.println(line);
  21.         } catch (IOException e) {
  22.             e.printStackTrace();
  23.         }
  24.     }
  25. }

2.Non Blocking IO（非阻塞）

  1. package com.lx.netty.nio;
  2.  
  3.  
  4. import java.io.IOException;
  5. import java.net.InetSocketAddress;
  6. import java.nio.ByteBuffer;
  7. import java.nio.channels.ServerSocketChannel;
  8. import java.nio.channels.SocketChannel;
  9. import java.util.ArrayList;
  10. import java.util.List;
  11.  
  12. public class NoBlockingServer {
  13.  
  14.     public static List<SocketChannel> channelList = new ArrayList<>();
  15.  
  16.     public static void main(String[] args) {
  17.  
  18.         try {
  19.             // 相当于serverSocket
  20.             // 1.支持非阻塞  2.数据总是写入buffer,读取也是从buffer中去读  3.可以同时读写
  21.             ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
  22.             serverSocketChannel.configureBlocking(false);
  23.             serverSocketChannel.socket().bind(new InetSocketAddress(8080));
  24.             while (true){
  25.                 // 这里将不再阻塞
  26.                 SocketChannel socketChannel = serverSocketChannel.accept();
  27.  
  28.                 if(socketChannel != null){
  29.                     socketChannel.configureBlocking(false);
  30.                     channelList.add(socketChannel);
  31.                 }else {
  32.                     System.out.println("没有请求过来！！！");
  33.                 }
  34.                 for (SocketChannel client : channelList){
  35.                     ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
  36.                     // 也不阻塞
  37.                     int num = client.read(byteBuffer);
  38.                     if(num>0){
  39.                         System.out.println("客户端端口："+ client.socket().getPort()+",客户端收据："+new String(byteBuffer.array()));
  40.                     }else {
  41.                         System.out.println("等待客户端写数据");
  42.                     }
  43.                 }
  44.             }
  45.  
  46.  
  47.         } catch (IOException e) {
  48.             e.printStackTrace();
  49.         }
  50.     }
  51. }

3.NIO

  1. package com.lx.netty.nio;
  2.  
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.SelectionKey;
  7. import java.nio.channels.Selector;
  8. import java.nio.channels.ServerSocketChannel;
  9. import java.nio.channels.SocketChannel;
  10. import java.util.Iterator;
  11. import java.util.Set;
  12.  
  13. public class NewIOServer {
  14.  
  15.     static Selector selector;
  16.  
  17.     public static void main(String[] args) {
  18.  
  19.         try {
  20.             selector = Selector.open();
  21.  
  22.             ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
  23.             serverSocketChannel.configureBlocking(false);
  24.             serverSocketChannel.socket().bind(new InetSocketAddress(8080));
  25.  
  26.             // 需要把serverSocketChannel注册到多路复用器上
  27.             serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
  28.  
  29.             while (true) {
  30.                 // 阻塞
  31.                 selector.select();
  32.                 Set<SelectionKey> selectionKeys = selector.selectedKeys();
  33.                 Iterator<SelectionKey> iterator = selectionKeys.iterator();
  34.                 while (iterator.hasNext()) {
  35.                     SelectionKey key = iterator.next();
  36.                     iterator.remove();
  37.                     if (key.isAcceptable()) {
  38.                         handlerAccept(key);
  39.                     } else if (key.isReadable()) {
  40.                         handlerRead(key);
  41.                     }else if(key.isWritable()){
  42.  
  43.                     }
  44.                 }
  45.  
  46.             }
  47.  
  48.         } catch (IOException e) {
  49.             e.printStackTrace();
  50.         }
  51.     }
  52.  
  53.     private static void handlerRead(SelectionKey key) {
  54.         SocketChannel socketChannel = (SocketChannel) key.channel();
  55.         ByteBuffer allocate = ByteBuffer.allocate(1024);
  56.         try {
  57.             socketChannel.read(allocate);
  58.             System.out.println("server msg:" + new String(allocate.array()));
  59.         } catch (IOException e) {
  60.             e.printStackTrace();
  61.         }
  62.  
  63.     }
  64.  
  65.     private static void handlerAccept(SelectionKey key) {
  66.         ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
  67.         // 不阻塞
  68.         try {
  69.             SocketChannel socketChannel = serverSocketChannel.accept();
  70.             socketChannel.configureBlocking(false);
  71.             socketChannel.write(ByteBuffer.wrap("I am wentai,is shuai".getBytes()));
  72.             // 读取客户端的数据
  73.             socketChannel.register(selector, SelectionKey.OP_READ);
  74.         } catch (IOException e) {
  75.             e.printStackTrace();
  76.         }
  77.  
  78.     }
  79.  
  80. }
  1. package com.lx.netty.nio;
  2.  
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.SelectionKey;
  7. import java.nio.channels.Selector;
  8. import java.nio.channels.SocketChannel;
  9. import java.util.Iterator;
  10. import java.util.Set;
  11.  
  12. public class NewIOClient {
  13.  
  14.     static Selector selector;
  15.  
  16.     public static void main(String[] args) {
  17.         try {
  18.             selector = Selector.open();
  19.             SocketChannel socketChannel = SocketChannel.open();
  20.             socketChannel.configureBlocking(false);
  21.             socketChannel.connect(new InetSocketAddress("localhost", 8080));
  22.  
  23.             // 需要把socketChannel注册到多路复用器上
  24.             socketChannel.register(selector, SelectionKey.OP_CONNECT);
  25.             while (true) {
  26.                 // 阻塞
  27.                 selector.select();
  28.                 Set<SelectionKey> selectionKeys = selector.selectedKeys();
  29.                 Iterator<SelectionKey> iterator = selectionKeys.iterator();
  30.                 while (iterator.hasNext()) {
  31.                     SelectionKey key = iterator.next();
  32.                     iterator.remove();
  33.                     if (key.isConnectable()) {
  34.                         handlerConnect(key);
  35.                     } else if (key.isReadable()) {
  36.                         handlerRead(key);
  37.                     } else if (key.isWritable()) {
  38.  
  39.                     }
  40.                 }
  41.  
  42.             }
  43.         } catch (IOException e) {
  44.             e.printStackTrace();
  45.         }
  46.  
  47.     }
  48.  
  49.     private static void handlerRead(SelectionKey key) {
  50.         SocketChannel socketChannel = (SocketChannel) key.channel();
  51.         ByteBuffer allocate = ByteBuffer.allocate(1024);
  52.         try {
  53.             socketChannel.read(allocate);
  54.             System.out.println("client msg:" + new String(allocate.array()));
  55.         } catch (IOException e) {
  56.             e.printStackTrace();
  57.         }
  58.  
  59.     }
  60.  
  61.     private static void handlerConnect(SelectionKey key) throws IOException {
  62.  
  63.         SocketChannel socketChannel = (SocketChannel) key.channel();
  64.         if (socketChannel.isConnectionPending()) {
  65.             socketChannel.finishConnect();
  66.         }
  67.         socketChannel.configureBlocking(false);
  68.         socketChannel.write(ByteBuffer.wrap("client I am wentai,is shuai".getBytes()));
  69.         socketChannel.register(selector,SelectionKey.OP_READ);
  70.     }
  71. }

4.AIO（异步IO）

  1. package com.lx.netty.aio;
  2.  
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.AsynchronousServerSocketChannel;
  7. import java.nio.channels.AsynchronousSocketChannel;
  8. import java.nio.channels.CompletionHandler;
  9.  
  10. // 服务端
  11. public class AIOServer {
  12.  
  13.     public static void main(String[] args) throws Exception {
  14.         // 创建一个SocketChannel并绑定了8080端口
  15.         final AsynchronousServerSocketChannel serverChannel =
  16.                 AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(8080));
  17.  
  18.         serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
  19.             @Override
  20.             public void completed(AsynchronousSocketChannel socketChannel, Object attachment) {
  21.                 try {
  22.                     // 打印线程的名字
  23.                     System.out.println("2--"+Thread.currentThread().getName());
  24.                     System.out.println(socketChannel.getRemoteAddress());
  25.                     ByteBuffer buffer = ByteBuffer.allocate(1024);
  26.                     // socketChannel异步的读取数据到buffer中
  27.                     socketChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
  28.                         @Override
  29.                         public void completed(Integer result, ByteBuffer buffer) {
  30.                             // 打印线程的名字
  31.                             System.out.println("3--"+Thread.currentThread().getName());
  32.                             buffer.flip();
  33.                             System.out.println(new String(buffer.array(), 0, result));
  34.                             socketChannel.write(ByteBuffer.wrap("HelloClient".getBytes()));
  35.                         }
  36.  
  37.                         @Override
  38.                         public void failed(Throwable exc, ByteBuffer buffer) {
  39.                             exc.printStackTrace();
  40.                         }
  41.                     });
  42.                 } catch (IOException e) {
  43.                     e.printStackTrace();
  44.                 }
  45.             }
  46.  
  47.             @Override
  48.             public void failed(Throwable exc, Object attachment) {
  49.                 exc.printStackTrace();
  50.             }
  51.         });
  52.  
  53.         System.out.println("1--"+Thread.currentThread().getName());
  54.         Thread.sleep(Integer.MAX_VALUE);
  55.     }
  56. }
  1. package com.lx.netty.aio;
  2.  
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.AsynchronousSocketChannel;
  7. import java.nio.channels.CompletionHandler;
  8.  
  9. public class AIOClient {
  10.  
  11.     private final AsynchronousSocketChannel client;
  12.  
  13.     public AIOClient() throws IOException {
  14.         client = AsynchronousSocketChannel.open();
  15.     }
  16.  
  17.     public static void main(String[] args) throws Exception {
  18.         new AIOClient().connect("localhost",8080);
  19.     }
  20.  
  21.     public void connect(String host, int port) throws Exception {
  22.         // 客户端向服务端发起连接
  23.         client.connect(new InetSocketAddress(host, port), null, new CompletionHandler<Void, Object>() {
  24.             @Override
  25.             public void completed(Void result, Object attachment) {
  26.                 try {
  27.                     client.write(ByteBuffer.wrap("这是一条测试数据".getBytes())).get();
  28.                     System.out.println("已发送到服务端");
  29.                 } catch (Exception e) {
  30.                     e.printStackTrace();
  31.                 }
  32.             }
  33.  
  34.             @Override
  35.             public void failed(Throwable exc, Object attachment) {
  36.                 exc.printStackTrace();
  37.             }
  38.         });
  39.  
  40.         final ByteBuffer bb = ByteBuffer.allocate(1024);
  41.         // 客户端接收服务端的数据，获取的数据写入到bb中
  42.         client.read(bb, null, new CompletionHandler<Integer, Object>() {
  43.             @Override
  44.             public void completed(Integer result, Object attachment) {
  45.                 // 服务端返回数据的长度result
  46.                 System.out.println("I/O操作完成：" + result);
  47.                 System.out.println("获取反馈结果：" + new String(bb.array()));
  48.             }
  49.  
  50.             @Override
  51.             public void failed(Throwable exc, Object attachment) {
  52.                 exc.printStackTrace();
  53.             }
  54.         });
  55.  
  56.         try {
  57.             Thread.sleep(Integer.MAX_VALUE);
  58.         } catch (InterruptedException e) {
  59.             e.printStackTrace();
  60.         }
  61.     }
  62.  
  63.  
  64. }

5.多路复用IO：多路复用有三种模型，一种时select()模型，一种poll模型，还一种就是epoll模型。

6.select()模型: 在这个socket中，每个socket都是有接收队列，发送队列，等待队列组成，所以每次调 用selector.select()方法的时候，都会把线程的引用放入到所有socket中的等待队列中，当接收到客户端的数 据后，把每一个socket上的等待队列中的线程移除，并放入到就绪队列中，然后去遍历文件列表，找出所有 接收到数据的socket。

select()的缺点：

1. 每次调用select都需要将线程加入到所有socket对象的等待队列中，每次唤醒进程又要将线程从所有socket对象的等待队列中移除。这里涉及到对socket列表的两次遍历，而且每次都要将整个fds列表传 递给内核，有一定的开销。正因为遍历操作开销大，出于效率的考量，才会规定select的最大监视数 量，默认只能监视1024个socket（强行修改也是可以的）；

2. 进程被唤醒后，程序并不知道socket列表中的那些socket上收到数据，因此在用户空间内需要对socket列表再做一次遍历。poll模型和select相似，只是对监听的socket没有限制。 我们发现其实select模型和poll模型其实每次都是 遍历所有的socket,有些socket其实没有事件，还是回去遍历，如果socket越多，那么遍历事件就越长，在高 并发的情况下，select模型的效率其实比较低，那么有没有一种模型，可以只返回有事件的socket呢，而不 需要遍历那么多的socket，答案就是epoll模型

7.epoll模型

其实epoll有两种工作方式，一种时LT，另外一种时ET。

LT模式意思就是通过是支持阻塞和非阻塞的socket,其实select和poll都是这种模式，我们今天主要谈的时ET模式，

ET模式只支持non-block socket，我们来看下epoll再ET模式下的工作原理。

1.NIO中调用Selector.open()得到一个selector,实际上就是调用底层的epoll_create函数，创建一个     eventpoll的事件文件对象，假设为ep_fd

2.NIO中调用register(selector, SelectionKey.OP_ACCEPT)，把需要注册的fd存储在jvm内存

3.循环调用selector.select()，底层触发epoll_ctl(ep_fd,add,fd,OP_ACCEPT),把需要注册的fd放到ep_fd对应 的红黑树上，注册OP_ACCEPT事件，并且对应到eventpoll文件对象中的rbr索引树中，然后调用epoll_wait()方法，然后把等待的线程的引用加入的eventpoll的等待队列中

4.一当接收到客户端的连接，那么将会触发socket上面的callback函数，并且把有数据处理的socket放入到eventpoll中的rdlist就序列表，同时唤醒等待队列中的线程，然后线程直接循环就绪列表中准备好的socket。

所以epoll模型的优点是：

1.支持一个进程打开很大数目的socket描述符

2.IO效率不随FD数目增加而线性下降

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/喵喵爱编程/article/detail/973947
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号