Netty

1.介绍

Netty是由JBOSS提供的一个java开源框架，现为 Github上的独立项目。Netty是一个异步的，基于事件驱动开发的网络应用框架用于开发高性能，高可靠性的网络IO程序。

也就是说，Netty 是一个基于NIO的客户、服务器端的编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户、服务端应用。Netty相当于简化和流线化了网络应用的编程开发过程，例如：基于TCP和UDP的socket服务开发。

2.BIO

BIO是(Block IO)即阻塞IO，是java最初的IO模型，JDK1.4之前唯一选择。

BIO中主要角色就是Server(可以自己写Client)，必须使用多线程。每次有客户端连接都要新建一个线程，去单独处理这个线程。

Server

  1. package bio;
  3. import java.io.IOException;
  4. import java.net.ServerSocket;
  5. import java.net.Socket;
  7. public class Server {
  8.     public static void main(String[] args) throws IOException {
  9.         ServerSocket serverSocket = null;
  10.         //服务端绑定端口
  11.         serverSocket = new ServerSocket(9000);
  12.         System.out.println("等待连接......");
  13.         //阻塞接收客户端
  14.         while (true){
  15.             Socket client = serverSocket.accept();
  16.             //直接创建线程
  17.             new Thread(new Runnable() {
  18.                 @Override
  19.                 public void run() {
  20.                     //System.out.println("连接成功");
  21.                     String msg = null;
  22.                     try {
  23.                         msg = handle(client);
  24.                     } catch (IOException e) {
  25.                         e.printStackTrace();
  26.                     }
  27.                     System.out.println("接收到的数据:"+msg);
  28.                 }
  29.             }).start();
  30.         }
  31.     }
  33.     public static String handle(Socket clientSocket) throws IOException {
  34.         byte buffer[] = new byte[1024];
  35.         String msg = null;
  36.         int len = clientSocket.getInputStream().read(buffer);
  37.         if(len != -1){
  38.             msg = new String(buffer,0,len,"utf-8");
  39.         }
  40.         return msg;
  41.     }
  42. }

Client

  1. package bio;
  3. import java.io.IOException;
  4. import java.io.OutputStream;
  5. import java.net.Socket;
  7. public class Client {
  8.     public void sendMsg(String name){
  9.         Socket socket = null;
  10.         try {
  11.             socket = new Socket("localhost",9000);
  12.             String msg = "这是来自"+name+"的消息";
  13.             byte[] buffer = msg.getBytes("utf-8");
  14.             OutputStream outputStream = socket.getOutputStream();
  15.             outputStream.write(buffer);
  16.             outputStream.close();
  17.             socket.close();
  18.         } catch (IOException e) {
  19.             e.printStackTrace();
  20.         }
  21.     }
  22. }

建立10个客户端通信

  1. package bio;
  3. import java.util.Random;
  4. import java.util.concurrent.ExecutorService;
  5. import java.util.concurrent.Executors;
  7. public class Clients {
  8.     public static void main(String[] args) throws InterruptedException {
  9. //        for (int i = 0; i < 5; i++) {
  10. //            //Thread.sleep(1000);
  11. //            int index = i;
  12. //            new Thread(new Runnable() {
  13. //                @Override
  14. //                public void run() {
  15. //                    new Client().sendMsg("xiao"+index);
  16. //                }
  17. //            }).start();
  18. //        }
  20.         /**
  21.          * 线程池版本
  22.          */
  23.         ExecutorService executorService = Executors.newFixedThreadPool(10);
  24.         for (int i = 0; i < 10; i++) {
  25.             executorService.submit(new Runnable() {
  26.                 @Override
  27.                 public void run() {
  28.                     new Client().sendMsg("xiao"+new Random().nextInt(99));
  29.                 }
  30.             });
  31.         }
  32.     }
  33. }

3.NIO

3.1.组件关系

  • 每个 channel 都会对应一个 Buffer
  • 一个线程对应一个Selector ， 一个Selector对应多个 channel(连接)
  • 程序切换到哪个 channel 是由事件决定的
  • Selector 会根据不同的事件，在各个通道上切换
  • Buffer 就是一个内存块 ， 底层是一个数组
  • 数据的读取写入是通过 Buffer完成的 ，BIO 中要么是输入流，或者是输出流, 不能双向，但是 NIO 的 Buffer 是可以读也可以写。
  • Java NIO系统的核心在于：通道(Channel)和缓冲区 (Buffer)。通道表示打开到 IO 设备(例如：文件、 套接字)的连接。若需要使用 NIO 系统，需要获取用于连接 IO 设备的通道以及用于容纳数据的缓冲区。然后操作缓冲区，对数据进行处理。简而言之，Channel 负责传输， Buffer 负责存取数据

3.2.Buffer

缓冲区Buffer，实质上是一个可以读写数据的内存块，可以理解为是一个容器对象(含数组)，该对象提供一组方法，可以轻松使用内存块，缓冲区对选哪个内置了一些机制，能够跟踪和记录缓冲区的状态变化，Channel提供从文件，网络读取数据的渠道，但是读取或写入的数据必须经由Buffer

Buffer4个属性

  1.     private int mark = -1;
  2.     private int position = 0;
  3.     private int limit;
  4.     private int capacity;
属性 描述
mark 标记
position 位置，下一个要读或写的元素的索引，每次读写缓冲区都会改变当前值，以便下次读写。
limit 表示缓冲区的当前终点，不能对缓冲区超过极限的位置进行读写，且极限是可以修改的。
Capacity 当前缓冲区的容量，在缓冲区创建时确定并不能修改。

0<=mark<=position<=limit<=capacity

  1. package nioBasic;
  3. import java.nio.IntBuffer;
  5. public class Buffer1 {
  6.     public static void main(String[] args) {
  7.         //声明一个10个大小的intBuffer
  8.         IntBuffer intBuffer = IntBuffer.allocate(10);
  9.         for (int i = 0; i < 10; i++) {
  10.             //在intBuffer中放入元素
  11.             intBuffer.put(i*10);
  12.         }
  13.         /**
  14.          public final Buffer flip() {
  15.             limit = position;
  16.             position = 0;
  17.             mark = -1;
  18.             return this;
  19.          }
  20.          在一次完整的读或写操作后position == limit，需要归零，
  21.          注意不是回到最初的状态limit == position[而是limit等于存放的数据的长度]
  22.          */
  23.         intBuffer.flip();
  24.         while (intBuffer.hasRemaining()/*判断当前position<limit*/){
  25.             System.out.println(intBuffer.get());
  26.         }
  27.     }
  28. }

3.3.Channel

Channel 经常翻译为通道，类似 IO 中的流，用于读取和写入。它与前面介绍的 Buffer 打交道，读操作的时候将 Channel 中的数据填充到 Buffer 中，而写操作时将 Buffer 中的数据写入到 Channel 中

  • FileChannel用于对文件的数据读写
  • SocketChannel与ServerSocketChannel对应Socket与ServerSocket，用于TCP通信
  • DatagramChannel用于UDP通信

FileChannel实验

实验1：通过nio向文件中输出内容

  1. package nioBasic;
  3. /**
  4.  * 通过nio向文件中输出内容
  5.  */
  7. import java.io.File;
  8. import java.io.FileOutputStream;
  9. import java.io.IOException;
  10. import java.nio.ByteBuffer;
  11. import java.nio.channels.FileChannel;
  13. public class Channel1 {
  14.     public static void main(String[] args) throws IOException {
  15.         //创建一个测试文件
  16.         File file = new File("nioTest1.txt");
  17.         if (!file.exists()){
  18.             file.createNewFile();
  19.         }
  20.         //获取文件输出流
  21.         FileOutputStream fos = new FileOutputStream(file);
  22.         //获取文件输出流的Channel
  23.         FileChannel channel = fos.getChannel();
  24.         //创建一个Buffer
  25.         ByteBuffer msg = ByteBuffer.wrap("Hello，肖云飞\n".getBytes());
  26.         //将Buffer的内容写到Channel中，由于这个Channel与指定文件绑定，所以会写到文件中
  27.         channel.write(msg);
  28.         fos.close();
  29.     }
  30. }

实验二乐观版本:

  1. package nioBasic;
  3. import java.io.FileInputStream;
  4. import java.io.IOException;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.FileChannel;
  8. public class Channel2_1 {
  9.     public static void main(String[] args) throws IOException {
  10.         FileInputStream fis = new FileInputStream("nioTest1.txt");
  11.         FileChannel channel = fis.getChannel();
  12.         /**
  13.          * 实现预知到文件内容小于1024字节
  14.          */
  15.         ByteBuffer buffer = ByteBuffer.allocate(1024);
  16.         int read = channel.read(buffer);
  17.         String msg = new String(buffer.array());
  18.         System.out.println(msg);
  19.         fis.close();
  20.     }
  21. }

实验二悲观版本

  1. package nioBasic;
  3. /**
  4.  * 通过nio从文件中获取内容
  5.  */
  7. import java.io.FileInputStream;
  8. import java.io.IOException;
  9. import java.nio.ByteBuffer;
  10. import java.nio.channels.FileChannel;
  11. import java.util.ArrayList;
  12. import java.util.List;
  14. public class Channel2_2 {
  15.     public static void main(String[] args) throws IOException {
  16.         FileInputStream fis = new FileInputStream("nioTest1.txt");
  17.         FileChannel channel = fis.getChannel();
  18.         //初始化一个大小为10的ByteBuffer
  19.         /**
  20.          * 注意，这个FileInputStream内置的Channel(FileChannel)是特殊的，只能使用ByteBuffer
  21.          * public abstract class FileChannel
  22.          *     extends AbstractInterruptibleChannel
  23.          *     implements SeekableByteChannel, GatheringByteChannel, ScatteringByteChannel{...}
  24.          */
  25.         ByteBuffer buffer = ByteBuffer.allocate(10);
  26.         List<Byte> bytes = new ArrayList<>();
  27.         while (true){
  28.             /**
  29.              * 注意：当buffer不能一次把所有的内容读完必须要有这句话
  30.              * 如果没有，第一次读完后position == limit，到极限了，之后每次都会读到0个，就会死循环
  31.              **/
  32.             buffer.clear();
  33.             int read = channel.read(buffer);
  34.             if(read == -1){break;}
  35.             else{
  36.                 /**
  37.                  * 现在我们无法预测文件的大小，所以不能确保Buffer的大小，
  38.                  * 所以“悲观地”认为Buffer无法一次读完，这种情况下需要拼接
  39.                  *
  40.                  * 但是有个问题，我们每次读可以得到一个ByteBuffer，里面是n个Byte，
  41.                  * 使用StringBuilder拼接的话append方法不能传入byte类型，只能是Char
  42.                  * 但如果使用char的话极有可能使中文字符(占三个字节)被分割开，出现乱码，
  43.                  * 所以需要将所有的ByteBuffer转化为byte[]后拼接为一个大的byte[],再转化为String
  44.                  */
  45.                 byte[] array = buffer.array();
  46.                 for (int i = 0; i < read; i++) {
  47.                     bytes.add(array[i]);
  48.                 }
  49.             }
  50.         }
  51.         /**
  52.          * 将List<byte>转化为byte[]，方便转化为字符串
  53.          */
  54.         byte[] msgArray = new byte[bytes.size()];
  55.         for (int i = 0; i < bytes.size(); i++) {
  56.             msgArray[i] = bytes.get(i);
  57.         }
  58.         System.out.println(new String(msgArray));
  59.         fis.close();
  60.     }
  61. }

MappedByteBuffer

MappedByteBuffer是ByteBuffer的子类，特点是首先它是针对RandomAccessFile使用的，所以是内存直接映射文件的，其次，它对RandomAccessFile功能进一步升级，可以指定映射的范围

  1. package nioBasic;
  3. import java.io.IOException;
  4. import java.io.RandomAccessFile;
  5. import java.nio.MappedByteBuffer;
  6. import java.nio.channels.FileChannel;
  8. public class MappedByteBufferTest {
  9.     public static void main(String[] args) throws IOException {
  10.         /**
  11.          * 创建随机读写文件(这个类只能操作文件)
  12.          * 随机读写文件的特点是可以把文件的操作像操作数组一样，是以byte为单位的
  13.          * 有一个特点是seek(long pos)方法，可以把文件的指针直接指向pos索引处开始操作
  14.          */
  15.         RandomAccessFile raf = new RandomAccessFile("nioTest1.txt","rw");
  16.         //获取channel
  17.         FileChannel channel = raf.getChannel();
  18.         /**
  19.          * public abstract MappedByteBuffer map(MapMode mode,long position, long size)
  20.          * 1.MapMode，这个Channel的操作模式（只读，读写，私有->写时复制）
  21.          * 2.position，从哪个位置开始映射
  22.          * 3.size，映射多少个块[多少个byte]
  23.          *
  24.          * 这就是MappedByteBuffer的特点，可以只将文件中某一段内容映射到内存中操作
  25.          */
  26.         MappedByteBuffer map = channel.map(FileChannel.MapMode.READ_WRITE, 0, 5);
  27.         //先可以将所有的映射读取出来
  28.         while (map.hasRemaining()){
  29.             System.out.print((char)map.get());
  30.         }
  31.         //map.flip();
  32.         System.out.println("\n========================");
  33.         //修改内容
  34.         map.put(0,(byte)'H');
  35.         map.put(1,(byte)'e');
  36.         map.put(2,(byte)'l');
  37.         map.put(3,(byte)'l');
  38.         map.put(4,(byte)'o');
  39.         //map.put(5,(byte)'p');会抛出异常，因为size是大小,不是索引
  41.         map.flip();
  42.         while (map.hasRemaining()){
  43.             System.out.print((char)map.get());
  44.         }
  45.         raf.close();
  46.     }
  47. }
  49. /**结果
  50. hELLO
  51. ========================
  52. Hello
  53. **/

3.4.Buffer的聚合与分散

分散/聚集 I/O 是使用多个而不是单个缓冲区来保存数据的读写方法。

一个分散的读取就像一个常规通道读取，只不过它是将数据读到一个缓冲区数组中而不是读到单个缓冲区中。同样地，一个聚集写入是向缓冲区数组而不是向单个缓冲区写入数据。

分散/聚集 I/O 对于将数据流划分为单独的部分很有用，这有助于实现复杂的数据格式。

ScatteringByteChannel、GatheringByteChannel 是两个interface，而 FileChannel 和 SocketChannel 等实现了这两个接口。

ScatteringByteChannel 是一个具有两个附加读方法的通道：

  1. long read( ByteBuffer[] dsts );
  2. long read( ByteBuffer[] dsts, int offset, int length );

这些 long read() 方法很像标准的 read 方法，只不过它们不是取单个缓冲区而是取一个缓冲区数组。

在 分散读取 中，通道依次填充每个缓冲区。填满一个缓冲区后，它就开始填充下一个,在移动下一个buffer前，必须填满当前的buffer，这也意味着它不适用于动态消息(消息大小不固定)。在某种意义上，缓冲区数组就像一个大缓冲区。

GatheringByteChannel聚集写入 类似于分散读取，只不过是用来写入。它也有接受缓冲区数组的方法：

  1. long write( ByteBuffer[] srcs );
  2. long write( ByteBuffer[] srcs, int offset, int length );

分散/聚集 I/O 对于将数据划分为几个部分很有用。例如，可能在编写一个使用消息对象的网络应用程序，每一个消息被划分为固定长度的头部和固定长度的正文。可以创建一个刚好可以容纳头部的缓冲区和另一个刚好可以容难正文的缓冲区。当您将它们放入一个数组中并使用分散读取来向它们读入消息时，头部和正文将整齐地划分到这两个缓冲区中。我们从缓冲区所得到的方便性对于缓冲区数组同样有效。因为每一个缓冲区都跟踪自己还可以接受多少数据，所以分散读取会自动找到有空间接受数据的第一个缓冲区。在这个缓冲区填满后，它就会移动到下一个缓冲区。聚集写对于把一组单独的缓冲区中组成单个数据流很有用。为了与上面的消息例子保持一致，可以使用聚集写入来自动将网络消息的各个部分组装为单个数据流，以便跨越网络传输消息。

3.5.Selector

选择器提供选择执行已经就绪的任务的能力。从底层来看，Selector提供了询问通道是否已经准备好执行每个I/O操作的能力。Selector 允许单线程处理多个Channel。仅用单个线程来处理多个Channels的好处是，只需要更少的线程来处理通道。事实上，可以只用一个线程处理所有的通道，这样会大量的减少线程之间上下文切换的开销。

并不是所有的Channel，都是可以被Selector 复用的。比方说，FileChannel就不能被选择器复用。

判断一个Channel 能被Selector 复用，有一个前提：判断他是否继承了一个抽象类SelectableChannel。如果继承了SelectableChannel，则可以被复用，否则不能。

SelectableChannel类提供了实现通道的可选择性所需要的公共方法。它是所有支持就绪检查的通道类的父类。所有socket通道，都继承了SelectableChannel类都是可选择的，包括从管道(Pipe)对象的中获得的通道。而FileChannel类，没有继承SelectableChannel，因此是不是可选通道。

3.6.nio服务端实例

  1. package nio;
  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;
  13. public class ServerPro {
  15.     public static void main(String[] args) throws IOException, InterruptedException {
  16.         //对服务端配置
  17.         ServerSocketChannel server = ServerSocketChannel.open();
  18.         server.bind(new InetSocketAddress(9000));
  19.         server.configureBlocking(false);
  20.         //开启多路复用器
  21.         Selector selector = Selector.open();
  22.         //将服务端Channel以OP_ACCEPT方式注册到复用器里面
  23.         server.register(selector, SelectionKey.OP_ACCEPT);
  24.         System.out.println("等待连接......");
  26.         while (true){
  27.             //多路复用器开始阻塞等待是否有事件发生(任何事件都会跳出阻塞)
  28.             selector.select();
  29.             Set<SelectionKey> selectionKeys = selector.selectedKeys();
  30.             Iterator<SelectionKey> channels = selectionKeys.iterator();
  31.             while(channels.hasNext()){
  32.             SelectionKey selectionKey = channels.next();
  33.             {
  34.                 //如果是接收事件，说明是有客户连接服务端
  35.                 if (selectionKey.isAcceptable()){
  36.                     ServerSocketChannel serverTmp = (ServerSocketChannel)selectionKey.channel();
  37.                     //获取客户通道，配置为非阻塞
  38.                     SocketChannel accept = serverTmp.accept();
  39.                     accept.configureBlocking(false);
  40.                     //因为建立连接，认为其会有读事件发生
  41.                     accept.register(selector,SelectionKey.OP_READ);
  42.                     System.out.println("连接成功......");
  43.                 }
  44.                 //如果是可读事件，说明是客户端
  45.                 else if (selectionKey.isReadable()){
  46.                     ByteBuffer buffer = ByteBuffer.allocate(256);
  47.                     SocketChannel clientTmp = (SocketChannel) selectionKey.channel();
  48.                     int len = clientTmp.read(buffer);
  49.                     if(len > 0){
  50.                         String msg = new String(buffer.array(),0,len);
  51.                         System.out.println("接收到的消息："+msg);
  52.                         if(msg.equals("通信完成，可以关闭")){
  53.                             if(!clientTmp.isConnected()){
  54.                                 System.out.println("异常结束");
  55.                             }
  56.                             else if(clientTmp.isConnected()){
  57.                                 selectionKey.cancel();
  58.                                 System.out.println("通信完成");
  59.                             }
  60.                         }
  61.                         else{
  62.                             clientTmp.register(selector,SelectionKey.OP_WRITE);
  63.                         }
  64.                     }
  65.                     else if(len == -1){
  66.                         System.out.println("断开连接");
  67.                         selectionKey.cancel();
  68.                     }
  69.                 }
  70.                 else if(selectionKey.isWritable()){
  71.                     SocketChannel  writeChannel = (SocketChannel)selectionKey.channel();
  72.                     writeChannel.write(ByteBuffer.wrap("已接收到客户端消息。".getBytes()));
  73.                     writeChannel.register(selector,SelectionKey.OP_READ);
  74.                     //Thread.sleep(2000);
  75.                 }
  76.                 channels.remove();
  77.             }
  78.         }
  79.         }
  80.     }
  81. }

3.7.nio客户端实例

  1. package nio;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.*;
  7. import java.util.Iterator;
  8. import java.util.Set;
  10. public class Client {
  12.     public void sendMsg(Selector selector,SocketChannel socket,String name) throws IOException, InterruptedException {
  13.         //SocketChannel socket = SocketChannel.open();
  14.         //Selector selector = Selector.open();
  15.         socket.configureBlocking(false);
  16.         socket.register(selector, SelectionKey.OP_CONNECT);
  17.         socket.connect(new InetSocketAddress("localhost",9000));
  19.         while(true){
  20.             selector.select();
  21.             Set<SelectionKey> selectionKeys = selector.selectedKeys();
  22.             Iterator<SelectionKey> channels = selectionKeys.iterator();
  23.             while (channels.hasNext()){
  24.                 SelectionKey channel = channels.next();
  25.                 if(channel.isConnectable()){
  26.                     //while(!socket.connect(new InetSocketAddress("localhost",9000)));
  28.                     /**
  29.                      * 下面这个判断是判断连接是否正在进行,不是必要的
  30.                      * 但是这个循环是必要的，直到socket.finishConnect()返回true才说明连接成功
  31.                      * 当然，在socket.finishConnect()返回false的阶段，客户端可以做其他事
  32.                      */
  34.                     //if(socket.isConnectionPending()){
  35.                         while(!socket.finishConnect());
  36.                     //}
  37.                         //if(socket.finishConnect()){
  38.                             String msg = name + "的消息";
  39.                             byte[] bytes = msg.getBytes("utf-8");
  40.                             ByteBuffer buffer = ByteBuffer.wrap(bytes);
  41.                             socket.write(buffer);
  42.                             //channel.cancel();
  43.                             socket.register(selector,SelectionKey.OP_READ);
  44.                         //}
  45.                 }
  46.                 else if(channel.isReadable()){
  47.                     SocketChannel  serverTmp = (SocketChannel)channel.channel();
  48.                     ByteBuffer buffer = ByteBuffer.allocate(256);
  49.                     int len = serverTmp.read(buffer);
  50.                     if(len > 0){
  51.                         String msg = (new String(buffer.array(),0,len).split("。"))[0];
  52.                         //System.out.println(msg.length());
  53.                         //System.out.println("已接收到客户端消息".length());
  54.                         System.out.println(msg);
  55.                         if(msg.equals("已接收到客户端消息")){
  56.                             System.out.println("正在关闭......");
  57.                             serverTmp.write(ByteBuffer.wrap("通信完成，可以关闭".getBytes()));
  58.                             //这里休眠2秒是为了确保服务端能正常关闭通道
  59.                             Thread.sleep(2000);
  60.                             System.exit(0);
  61.                         }
  62.                     }
  63.                     else if (len == -1){
  64.                         System.out.println("连接断开");
  65.                         channel.cancel();
  66.                     }
  67.                 }
  68.                 channels.remove();
  69.             }
  70.         }
  71.     }
  72. }

3.8.多客户端运行(多线程)

  1. package nio;
  3. import java.io.IOException;
  4. import java.nio.channels.Selector;
  5. import java.nio.channels.SocketChannel;
  6. import java.util.concurrent.ExecutorService;
  7. import java.util.concurrent.Executors;
  9. public class Clients {
  10.     public static void main(String[] args) throws IOException {
  11.         ExecutorService executorService = Executors.newCachedThreadPool();
  12.         for (int i = 0; i < 10; i++) {
  13.             String id = new Integer(i).hashCode()+"";
  14.             executorService.execute(
  15.                     new Runnable() {
  16.                         @Override
  17.                         public void run() {
  18.                             try {
  19.                                 //让每个客户端channel都持有一个selector
  20.                                 SocketChannel channel = SocketChannel.open();
  21.                                 Selector selector = Selector.open();
  22.                                 new Client().sendMsg(selector,channel,"肖云飞"+id);
  23.                             } catch (IOException e) {
  24.                                 e.printStackTrace();
  25.                             } catch (InterruptedException e) {
  26.                                 e.printStackTrace();
  27.                             }
  28.                         }
  29.                     }
  30.             );
  31.         }
  32.     }
  33. }

3.9.小实战

nio实现群聊系统

服务端

  1. package nio.groupChat;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.*;
  7. import java.util.Iterator;
  8. import java.util.Set;
  10. public class ChatServer {
  11.     private Selector selector = null;
  12.     private ServerSocketChannel serverSocketChannel = null;
  13.     private final int port = 9000;
  15.     //初始化属性
  16.     public ChatServer(){
  17.         try {
  18.             selector = Selector.open();
  19.             serverSocketChannel = ServerSocketChannel.open();
  20.             serverSocketChannel.socket().bind(new InetSocketAddress(port));
  21.             serverSocketChannel.configureBlocking(false);
  22.             serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
  23.         } catch (IOException e) {
  24.             e.printStackTrace();
  25.         }
  26.     }
  27.     public void listen() throws IOException {
  28.         while (true){
  29.             int sum = selector.select(10000);
  30.             if(sum > 0){
  31.                 Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
  32.                 while (iterator.hasNext()){
  33.                     SelectionKey key = iterator.next();
  34.                     if(key.isAcceptable()){
  35.                         SocketChannel client = serverSocketChannel.accept();
  36.                         client.configureBlocking(false);
  37.                         client.register(selector,SelectionKey.OP_READ);
  38.                         System.out.println(client.getRemoteAddress()+"上线");
  39.                     }else if(key.isReadable()){
  40.                         SocketChannel client = (SocketChannel) key.channel();
  41.                         Boolean flag = true;
  42.                         //读取并广播
  43.                         readAndBroadcast(client,flag);
  44.                         System.out.println();
  45.                     }
  46.                     iterator.remove();
  47.                 }
  48.             }else{
  49.                 continue;
  50.                 //System.out.println("无连接");
  51.             }
  52.         }
  53.     }
  54.     /**
  55.      * 从指定客户端读取消息
  56.      * @param client 客户顿
  57.      * @param flag 标志，表示是否可以继续读
  58.      */
  59.     private void readAndBroadcast(SocketChannel client,Boolean flag){
  60.         while (flag) {
  61.             ByteBuffer buffer = ByteBuffer.allocate(1024);
  62.             String msg = null;
  63.             try {
  64.                 int read = client.read(buffer);
  65.                 if (read > 0) {
  66.                     msg = new String(buffer.array());
  67.                     System.out.println(client.getRemoteAddress()+"说:"+msg);
  68.                     //接收到消息后广播
  69.                     broadCast(msg,client);
  70.                 } else {
  71.                     //没有消息说明已经不用读了,但通道不用注销
  72.                     flag = false;
  73.                 }
  74.             } catch (IOException e) {
  75.                 try {
  76.                     System.out.println("客户端" + client.getRemoteAddress() + "已下线");
  77.                     //出现异常，关闭通道
  78.                     flag = false;
  79.                     client.close();
  80.                 } catch (IOException ex) {
  81.                     ex.printStackTrace();
  82.                 }
  83.             }
  84.         }
  85.     }
  87.     /**
  88.      * 广播消息
  89.      * @param msg 消息
  90.      * @param self 广播对象需要排除自己
  91.      */
  92.     private void broadCast(String msg,SocketChannel self) throws IOException {
  93.         /**
  94.          * 广播不需要selector.select()了，直接遍历所有注册到selector的key
  95.          * 遍历排除serverSocketChannel与自己
  96.          * 发送消息
  97.          */
  98.         Set<SelectionKey> keys = selector.keys();
  99.         for (SelectionKey key : keys) {
  100.             SelectableChannel channel = key.channel();
  101.             if (channel instanceof SocketChannel && channel != self) {
  102.                 SocketChannel client = (SocketChannel)channel;
  103.                 String endMag = "["+self.getRemoteAddress()+"]->"+msg;
  104.                 ByteBuffer buffer = ByteBuffer.wrap(endMag.getBytes());
  105.                 client.write(buffer);
  106.             }
  107.         }
  108.     }
  109.     public static void main(String[] args) throws IOException {
  110.         new ChatServer().listen();
  111.     }
  112. }

客户端

  1. package nio.groupChat;
  4. import java.io.IOException;
  5. import java.net.InetSocketAddress;
  6. import java.nio.ByteBuffer;
  7. import java.nio.channels.SelectionKey;
  8. import java.nio.channels.Selector;
  9. import java.nio.channels.SocketChannel;
  10. import java.util.Iterator;
  11. import java.util.Scanner;
  13. public class ChatClient {
  14.     private Selector selector = null;
  15.     private SocketChannel client = null;
  16.     private static final String RemoveHost = "localhost";
  17.     private static final int port = 9000;
  19.     //初始化
  20.     public ChatClient(){
  21.         try {
  22.             selector = Selector.open();
  23.             client = SocketChannel.open();
  24.             client.configureBlocking(false);
  25.             //初始注册为SelectionKey.OP_CONNECT
  26.             client.register(selector, SelectionKey.OP_CONNECT);
  27.             client.connect(new InetSocketAddress(RemoveHost,port));
  28.         } catch (IOException e) {
  29.             e.printStackTrace();
  30.         }
  31.     }
  33.     /**
  34.      * 确保客户端的连接成功
  35.      * @throws IOException
  36.      */
  37.     public void connect() throws IOException {
  38.         selector.select();
  39.         Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
  40.         while (iterator.hasNext()){
  41.             SelectionKey key = iterator.next();
  42.             if(key.isConnectable()){
  43.                 while (!client.finishConnect());
  44.                 client.register(selector,SelectionKey.OP_READ);
  45.             }
  46.         }
  47.     }
  48.     public void write() throws IOException {
  49.         //必须等到连接建立，如果先调用了上面的connect方法就不需要了
  50.         //while (!client.finishConnect());
  51.         /**
  52.          * 确定连接建立后，就把原来的CONNECT改注册为READ
  53.          */
  54.         //client.register(selector,SelectionKey.OP_READ);
  55.         //发消息
  56.         while (true){
  57.             Scanner scanner = new Scanner(System.in);
  58.             String msg = scanner.nextLine();
  59.             ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
  60.             try {
  61.                 client.write(buffer);
  62.             } catch (IOException e) {
  63.                 e.printStackTrace();
  64.             }
  65.         }
  66.     }
  67.     public void readInfo() throws IOException {
  68.         //必须等到连接建立，如果先调用了上面的connect方法就不需要了
  69.         //while (!client.finishConnect());
  70.         /**
  71.          * 确定连接建立后，就把原来的CONNECT改注册为READ
  72.          */
  73.         //client.register(selector,SelectionKey.OP_READ);
  74.         try {
  75.             while (true){
  76.                 int sum = selector.select(2000);
  77.                 if(sum > 0){
  78.                     Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
  79.                     while (iterator.hasNext()){
  80.                         SelectionKey key = iterator.next();
  81.                         if(key.isReadable()){
  82.                             ByteBuffer buffer = ByteBuffer.allocate(1024);
  83.                             int read = client.read(buffer);
  84.                             //System.out.print("接收到消息：");
  85.                             System.out.println(new String(buffer.array()));
  86.                         }
  87.                         iterator.remove();
  88.                     }
  89.                 }else {
  90.                     continue;
  91.                 }
  92.             }
  93.         } catch (IOException e) {
  94.             e.printStackTrace();
  95.         }
  96.     }
  97. }

客户端开通

  1. package nio.groupChat;
  3. import java.io.IOException;
  5. public class Client0 {
  6.     public static void main(String[] args) throws IOException {
  7.         ChatClient chatClient0 = new ChatClient();
  8.         //如果在write()与readInfo()里面加了确保连接以及修改注册的语句，就可以不用connect()了
  9.         /**
  10.          * 但是我发现由于初始注册的是CONNECT，就需要在建立连接后修改注册为READ,否则无法接受消息
  11.          * 如果不执行connect()，就需要在write()与readInfo()中都添加修改注册的语句
  12.          * (因为不能确保当前客户端先执行write（）)，这样会重复执行两次，不是很好
  13.          */
  14.         chatClient0.connect();
  15.         new Thread(new Runnable() {
  16.             @Override
  17.             public void run() {
  18.                 try {
  19.                     chatClient0.write();
  20.                 } catch (IOException e) {
  21.                     e.printStackTrace();
  22.                 }
  23.             }
  24.         }).start();
  25.         new Thread(new Runnable() {
  26.             @Override
  27.             public void run() {
  28.                 try {
  29.                     chatClient0.readInfo();
  30.                 } catch (IOException e) {
  31.                     e.printStackTrace();
  32.                 }
  33.             }
  34.         }).start();
  35.     }
  36. }
  1. package nio.groupChat;
  3. import java.io.IOException;
  5. public class Client1 {
  6.     public static void main(String[] args) throws IOException {
  7.         ChatClient chatClient1 = new ChatClient();
  8.         /**
  9.          * 确保连接建立
  10.          */
  11.         chatClient1.connect();
  12.         new Thread(new Runnable() {
  13.             @Override
  14.             public void run() {
  15.                 try {
  16.                     chatClient1.write();
  17.                 } catch (IOException e) {
  18.                     e.printStackTrace();
  19.                 }
  20.             }
  21.         }).start();
  23.         new Thread(new Runnable() {
  24.             @Override
  25.             public void run() {
  26.                 try {
  27.                     chatClient1.readInfo();
  28.                 } catch (IOException e) {
  29.                     e.printStackTrace();
  30.                 }
  31.             }
  32.         }).start();
  33.     }
  34. }
  1. package nio.groupChat;
  3. import java.io.IOException;
  5. public class Client2 {
  6.     public static void main(String[] args) throws IOException {
  7.         ChatClient chatClient2 = new ChatClient();
  8.         /**
  9.          * 确保连接建立
  10.          */
  11.         chatClient2.connect();
  12.         new Thread(new Runnable() {
  13.             @Override
  14.             public void run() {
  15.                 try {
  16.                     chatClient2.write();
  17.                 } catch (IOException e) {
  18.                     e.printStackTrace();
  19.                 }
  20.             }
  21.         }).start();
  23.         new Thread(new Runnable() {
  24.             @Override
  25.             public void run() {
  26.                 try {
  27.                     chatClient2.readInfo();
  28.                 } catch (IOException e) {
  29.                     e.printStackTrace();
  30.                 }
  31.             }
  32.         }).start();
  33.     }
  34. }

3.10.零拷贝

在java程序中，常用的零拷贝实现有mmap（内存映射）和sendFile

关于零拷贝，在另一个md里面,下面是零拷贝在nio中的使用

NewIOClient

  1. import java.io.FileInputStream;
  2. import java.io.IOException;
  3. import java.net.InetSocketAddress;
  4. import java.nio.channels.FileChannel;
  5. import java.nio.channels.SocketChannel;
  7. public class NewIOClient {
  8.     public static void main(String[] args) throws IOException {
  9.         SocketChannel socketChannel = SocketChannel.open();
  10.         socketChannel.connect(new InetSocketAddress("localhost",4399));
  11.         String filename="CodeForces.rar";
  13.         //得到文件的channel
  14.         FileChannel fileChannel = new FileInputStream(filename).getChannel();
  16.         long startTimeMillis = System.currentTimeMillis();
  18.         //在Linux下一个transto函数可以完成传输
  19.         //在windows下一次调用只能发送8M的文件,需要分段传输文件，而且要注意传送时的位置
  20.         //transferto底层用到零拷贝,传送到socketchannel中
  21.         long transfercount = fileChannel.transferTo(0, fileChannel.size(), socketChannel);
  22.         System.out.println("发送的字节数为:"+transfercount+"，耗时为:" + (System.currentTimeMillis() - startTimeMillis));
  23.         fileChannel.close();
  25.     }
  26. }

NewIOServer

  1. import java.io.IOException;
  2. import java.net.InetSocketAddress;
  3. import java.net.ServerSocket;
  4. import java.nio.ByteBuffer;
  5. import java.nio.channels.ServerSocketChannel;
  6. import java.nio.channels.SocketChannel;
  8. public class NewIOServer {
  9.     public static void main(String[] args) throws IOException {
  10.         InetSocketAddress inetSocketAddress = new InetSocketAddress(4399);
  11.         ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
  12.         ServerSocket serverSocket = serverSocketChannel.socket();
  13.         serverSocket.bind(inetSocketAddress);
  15.         ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
  16.         while (true){
  17.             SocketChannel socketChannel = serverSocketChannel.accept();
  18.             int readacount=0;
  19.             while (readacount != -1){
  20.                 try {
  21.                     readacount = socketChannel.read(byteBuffer);
  22.                 }catch (Exception e){
  23.                     break;
  24.                 }
  25.                 byteBuffer.rewind();    //将buffer的position置为0，mark作废
  26.             }
  27.         }
  28.     }
  29. }