Netty

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

package com.clover.bio;

import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BIOServer {
    public static void main(String[] args) throws IOException {
        // 线程池机制

        //思路
        //1.创建一个线程池
        //2.如果有客户端连接，就创建一个线程，与之通信(单独写一个方法)

        ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();

        // 创建ServerSocket
        ServerSocket serverSocket = new ServerSocket(6666);

        System.out.println("服务器启动了");

        while(true){
            // 监听，等待客户端连接
            System.out.println("线程信息 id =" + Thread.currentThread().getId() + "名字=" + Thread.currentThread().getName());
            System.out.println("等待连接.........");
            final Socket socket = serverSocket.accept();
            System.out.println("连接到一个客户端");

            // 连接到一个客户端就创建一个线程，与之通信(单独写一个方法)
            newCachedThreadPool.execute(new Runnable() {
                // 我们可以重写这个方法
                @Override
                public void run() {
                    // 可以和客户端通讯
                    handler(socket);
                }
            });
        }
    }

    // 编写一个handler方法和客户端通讯
    public static void handler(Socket socket){

        try {
            System.out.println("线程信息 id =" + Thread.currentThread().getId() + "名字=" + Thread.currentThread().getName());
            byte[] bytes = new byte[1024];

            // 通过socket获取输入流
            InputStream inputStream = socket.getInputStream();

            // 循环读取客户端发送的数据
            while(true){
                System.out.println("线程信息 id =" + Thread.currentThread().getId() + "名字=" + Thread.currentThread().getName());

                System.out.println("read.........");
                int read = inputStream.read(bytes);
                if(read != -1){
                    // 不等于-1说明还未读完，还可以继续读取
                    // 输出客户端发送的数据
                    System.out.println(new java.lang.String(bytes,0,read));
                }else{
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.out.println("关闭和client的连接");
            try {
                socket.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

package com.clover.nio;

import java.io.FileOutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

public class NIOFileChannel01 {
  public static void main(String[] args) throws Exception{
      String str = "hello,韋";
      //创建一个输出流->然后获取内置channel
      FileOutputStream fileOutputStream = new FileOutputStream("H:\\file01.txt");

      //通过FileOutputStream获取对应的FileChannel
      //这个FileChannel真实类型为FileChannelImpl
      FileChannel fileChannel = fileOutputStream.getChannel();

      //创建一个缓冲区 ByteBuffer
      ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

      //将数据放入缓冲区中
      byteBuffer.put(str.getBytes());

      //数据写入完毕后要将buffer反转
      byteBuffer.flip();

      // 将缓冲区数据写入fileChannel
      fileChannel.write(byteBuffer);
      //因为fileChannel中封装了Java的输出流，最终写入文件中也是通过流写入，所以最后需要关闭
      fileOutputStream.close();
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

package com.clover.nio;

import java.io.File;
import java.io.FileInputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

public class NIOFileChannel02 {
  public static void main(String[] args) throws Exception {
      //创建一个文件的输入流->然后获取内置channel
      File file = new File("H:\\file01.txt");
      FileInputStream fileInputStream = new FileInputStream(file);

      //通过fileInputStream获取对应的channel
      FileChannel fileChannel = fileInputStream.getChannel();

      //创建一个缓冲区
      ByteBuffer byteBuffer = ByteBuffer.allocate((int) file.length());

      //将通道中的数据向缓冲区中写入
      fileChannel.read(byteBuffer);

      //将byteBuffer的字节数据转成String
      System.out.println(new String(byteBuffer.array()));

      fileInputStream.close();
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

package com.clover.nio;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

public class NIOFileChannel03 {
  public static void main(String[] args) throws Exception {
      // 创建一个文件
      File file = new File("H:\\1.txt");
      FileInputStream fileInputStream = new FileInputStream(file);
      // 通过文件输入流获取channel
      FileChannel fileInputStreamChannel = fileInputStream.getChannel();

      // 创建一个输出流，读取数据
      FileOutputStream fileOutputStream = new FileOutputStream("H:\\2.txt");
      // 获取对应的channel
      FileChannel fileOutputStreamChannel = fileOutputStream.getChannel();

      // 创建一个缓冲区
      ByteBuffer byteBuffer = ByteBuffer.allocate((int) file.length());

      while (true){
          // 这里有一个重要操作，一定不能忘记
          byteBuffer.clear(); // 清空buffer
          // 将通道数据写入缓冲区
          int read = fileInputStreamChannel.read(byteBuffer);
          if (read == -1){
              break;
          }
          // buffer写入完数据后要反转为后面从buffer读取数据做准备
          byteBuffer.flip();
          // 将缓冲区数据读取到通道中
          fileOutputStreamChannel.write(byteBuffer);
      }

      // 关闭相关的流
      fileInputStream.close();
      fileOutputStream.close();
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

package com.clover.nio;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.nio.channels.FileChannel;

public class NIOFileChannel04 {
  public static void main(String[] args) throws Exception {
      // 创建相关的流
      FileInputStream fileInputStream = new FileInputStream("H:\\1.jpg");
      FileOutputStream fileOutputStream = new FileOutputStream("H:\\2.jpg");

      // 获取相关流所对应的通道
      FileChannel fileInputStreamChannel = fileInputStream.getChannel();
      FileChannel fileOutputStreamChannel = fileOutputStream.getChannel();

      // 将文件从目标通道拷贝到当前通道
      fileOutputStreamChannel.transferFrom(fileInputStreamChannel,0,fileInputStreamChannel.size());

      // 关闭相关的通道和流
      fileInputStreamChannel.close();
      fileOutputStreamChannel.close();
      fileInputStream.close();
      fileOutputStream.close();
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

package com.clover.nio;

import java.nio.ByteBuffer;

public class NIOByteBufferPutGet {
  public static void main(String[] args) {
      // 创建一个Buffer
      ByteBuffer buffer = ByteBuffer.allocate(64);

      //类型化方式放入数据
      buffer.putInt(824);
      buffer.putLong(8);
      buffer.putChar('韋');
      buffer.putShort((short) 4);

      // 取出数据
      buffer.flip();

      System.out.println(buffer.getInt());
      System.out.println(buffer.getLong());
      System.out.println(buffer.getChar());
      System.out.println(buffer.getShort());
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

package com.clover.nio;

import java.nio.ByteBuffer;

public class ReadOnlyBuffer {
  public static void main(String[] args) {
      // 创建一个Buffer
      ByteBuffer buffer = ByteBuffer.allocate(64);
      for (int i= 0; i < 64; i++){
          buffer.put((byte) i);
      }

      //读取
      buffer.flip();

      // 得到一个只读的Buffer
      ByteBuffer readOnlyBuffer = buffer.asReadOnlyBuffer();
      System.out.println(readOnlyBuffer.getClass());

      while (readOnlyBuffer.hasRemaining()){
          // 这个方法会让我们的position自动加1
          System.out.println(readOnlyBuffer.get());
      }

      readOnlyBuffer.put((byte) 824); // 用只读的Buffer进行写入操作会抛出ReadOnlyBufferException
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

package com.clover.nio;

import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;

/*
 * 说明
 * 1.MappedByteBuffer 可以让文件直接在内存(堆外内存)中修改，操作系统不需要拷贝一次
 */

public class MappedByteBufferTest {
  public static void main(String[] args) throws Exception {
      RandomAccessFile randomAccessFile = new RandomAccessFile("H:\\1.txt", "rw");
      //获取对应的通道
      FileChannel channel = randomAccessFile.getChannel();

      /*
       * 参数1：FileChannel.MapMode.READ_WRITE 使用的读写模式
       * 参数2：0 ：可以直接修改的起始位置
       * 参数3：5 ：是映射到内存的大小(不是索引位置)，即将 1.txt的多少个字节映射到内存
       * 可以直接修改的范围就是0-5
       * MappedByteBuffer 实际类型 DirectByteBuffer
       */
      MappedByteBuffer mappedByteBuffer = channel.map(FileChannel.MapMode.READ_WRITE, 0, 5);

      mappedByteBuffer.put(0, (byte) 'H');
      mappedByteBuffer.put(3, (byte) '8');
      mappedByteBuffer.put(5, (byte) '8');  //这个地方的5并不是代表索引，而是代表可以修改字节数的大小，超过5会抛出异常IndexOutOfBoundsException

      randomAccessFile.close() ;

  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

package com.clover.nio;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Arrays;

/*
 * Scattering：将数据写入到buffer时，可以采用buffer数组，依次写入  [分散]
 * Gathering：从buffer读取数据时，可以采用buffer数组，依次读取
 */
public class ScatteringAndGatheringTest {
  public static void main(String[] args) throws Exception {
      // 使用ServerSocketChannel 和 SocketChannel  这就需要网络了
      ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
      InetSocketAddress inetSocketAddress = new InetSocketAddress(7000);

      // 绑定端口到socket，并启动
      serverSocketChannel.socket().bind(inetSocketAddress);

      // 创建buffer数组
      ByteBuffer[] byteBuffers = new ByteBuffer[2];
      byteBuffers[0] = ByteBuffer.allocate(5);
      byteBuffers[1] = ByteBuffer.allocate(3);

      // 等待客户端连接(telnet)
      SocketChannel socketChannel = serverSocketChannel.accept();

      int messageLength = 8;   // 假设我们从客户端最多接受8个字节

      // 循环读取
      while (true){
          // 统计读取了多少个字节
          int byteRead = 0;
          while (byteRead < messageLength){
              long l = socketChannel.read(byteBuffers);
              byteRead += l;// 累计读取的字节数
              System.out.println("byteRead =" + byteRead);
              // 使用流打印，查看当前buffer的position和limit
              Arrays.asList(byteBuffers).stream().map(buffer -> "position=" + buffer.position() + ",limit=" + buffer.limit()).forEach(System.out::println);
          }

          // 将所有的buffer进行一个反转flip
          Arrays.asList(byteBuffers).forEach(buffer -> buffer.flip());

          // 将数据读取出来显示在控制台
          long byteWrite = 0;// 查看显示在控制台字节大小
          while (byteWrite < messageLength){
              long w = socketChannel.write(byteBuffers);
              byteWrite += w;
          }

          // 将所有的buffer进行clear
          Arrays.asList(byteBuffers).forEach(buffer -> {
              // 如果数据是全部读取完毕，接着再次往buffer写数据，这时position是上次读取位置，limit是最大可读取位置，全部读取完成position=limit
              // 那么直接往里面写数据会报错，因为position必须小于limit，而我们也想从0开始存储，并且需要让最大存储限制limit应该是容量capacity
              // 那我们就需要手动修改position和limit的值。让position=0，而limit=capacity，buffer提供了这个方法clear()
              buffer.clear();
          });

          System.out.println("byteRead=" + byteRead + ",byteWrite=" + byteWrite + ",messageLength=" + messageLength);
      }
  }
}

1
2
3
4
5

public abstract class Selector implements Closeable{
		public static Selector open(); // 得到一个选择器对象
		public int select(long timeout); // 监控所有注册的通道，当其中有IO操作可以进行时，将对应的SelectionKey加入到内部集合中并返回，参数用来设置超时时间
		public Set<SelectionKey> selectedKeys(); // 从内部集合中得到所有的SelectionKey
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

package com.clover.nio;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

public class NIOServer {
  public static void main(String[] args) throws Exception{
      // 创建ServerSocketChannel -> ServerSocket
      ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

      // 得到一个Selector对象
      Selector selector = Selector.open();

      // 绑定一个端口6666，在服务器端监听
      serverSocketChannel.socket().bind(new InetSocketAddress(6666));

      // 设置为非阻塞
      serverSocketChannel.configureBlocking(false);

      // 把 serverSocketChannel 注册到 selector 关心事件为OP_ACCEPT
      serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

      // 循环等待客户端连接
      while (true){
          // 这里我们等待1秒，如果没有事件发生，就返回
          if (selector.select(1000) == 0){
              // 没有事件发生
              System.out.println("服务器等待了1秒，无连接");
              continue;
          }

          // 如果返回的 >0,就获取到相关的selectionKey集合
          // 1.如果返回的 >0,表示已经获取到关注的事件
          // 2.selector.selectedKeys() 返回关注事件的集合
          // 3.通过 selectionKeys 反向获取通道
          Set<SelectionKey> selectionKeys = selector.selectedKeys();

          // 遍历<SelectionKey> ，使用迭代器遍历
          Iterator<SelectionKey> keyIterator = selectionKeys.iterator();

          while (keyIterator.hasNext()){
              // 获取到SelectionKey
              SelectionKey key = keyIterator.next();

              //根据key对应的通道发生的事件做相应的处理
              if (key.isAcceptable()){
                  // 如果是 OP_ACCEPT，有新的客户端连接
                  // 给该客户端生成一个SocketChannel
                  // 这里使用 accept 为什么不会产生阻塞？ 因为你原本的判断就已经判定了它时处于哪一个事件，自然不会在去等待
                  SocketChannel socketChannel = serverSocketChannel.accept();
                  System.out.println("客户端连接成功  生成了一个 socketChannel " + socketChannel.hashCode());
                  socketChannel.configureBlocking(false);
                  // 将 socketChannel 注册到selector，关注事件为 OP_READ，同时给该 socketChannel 关联一个Buffer
                  /*
                   * 这里需要注意：
                   *    如果你在将 socketChannel 注册到selector时，给其关联一个buffer分配空间过大时，在
                   *    后面使用输出语句进行输出时，没有字节填充时，后面会默认输出乱码
                   */
                  socketChannel.register(selector,SelectionKey.OP_READ, ByteBuffer.allocate(9));
              }

              if (key.isReadable()){
                  // 发生 OP_READ 事件
                  // 通过key反向获取到对应的channel
                  SocketChannel channel = (SocketChannel)key.channel();

                  // 获取到该Channel关联的Buffer
                  ByteBuffer buffer = (ByteBuffer)key.attachment();

                  // 将通道中的数据读取到Buffer
                  channel.read(buffer);
                  System.out.println("从客户端发送的数据" + new String(buffer.array()));
              }

              // 手动从集合中移除当前的 SelectionKey ，防止重复操作
              keyIterator.remove();
          }
      }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

package com.clover.nio;

import jdk.nashorn.internal.ir.CallNode;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

public class NIOClient {
  public static void main(String[] args) throws Exception{
      // 得到一个网络通道
      SocketChannel socketChannel = SocketChannel.open();
      // 设置非阻塞
      socketChannel.configureBlocking(false);
      // 提供服务器端的ip 和 端口
      InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 6666);
      // 连接服务器
      if (!socketChannel.connect(inetSocketAddress)){
          while (!socketChannel.finishConnect()){
                System.out.println("因为连接需要事件，客户端不会阻塞，可以做其它工作");
          }
      }

      // 如果连接成功，就发送数据‘
      String str = "hello,韋";
      // 使用这种方式就不需要再去指定字节大小了，也不再浪费空间大小
      ByteBuffer buffer = ByteBuffer.wrap(str.getBytes());
      // 发送数据,将Buffer数据写入channel
      socketChannel.write(buffer);
      System.in.read();// 使代码停止在此
  }
}

1
2
3
4
5
6
7
8
9

public abstract class SelectionKey {
		public abstract Selector selector();// 得到与之关联的Selector对象
		public abstract SelectableChannel channel();//得到与之关联的通道
		public final Object attachment();//得到与之关联的共享数据
		public abstract SelectionKey interestOps(int ops);//设置或改变监听事件，比如：原本是监听连接事件，你可以将其改为监听读事件，这个根据需求来定
		public final boolean isAcceptable();//是否可以accept
		public final boolean isReadable();//是否可以读
		public final boolean isWriteable();//是否可以写
}

1
2
3
4
5
6
7

public abstract class ServerSocketChannel extends AbstractSelectableChannel implements NetworkChannel{
		public static ServerSocketChannel open();//得到一个ServerSocketChannel通道
		public final ServerSocketChannel bind(SocketAddress local);//设置服务器端端口号
		public final SelectableChannel configureBlocking(boolean block);//设置阻塞或非阻塞模式，取值false表示采用非阻塞模式
		public abstract SocketChannel accept();//接受一个连接，返回代表这个连接的通道对象
		public final SelectionKey register(Selector sel, int ops);//注册一个选择器并设置监听事件
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

public abstract class SocketChannel extends AbstractSelectableChannel implements ByteChannel, ScatteringByteChannel, GatheringByteChannel, NetworkChannel {
		public static SocketChannel open();//得到一个Socket Channel通道
		public fianl SelectableChannel configureBlocking(boolean block);// 设置阻塞或非阻塞模式，取值false表示采用非阻塞模式
		public abstract boolean connect(SocketAddress remote);//连接服务器
		public abstract boolean finishConnect();//如果上面的方法连接失败，接下来就要通过该方法完成连接操作
		public int write(ByteBuffer src);//往通道里面写数据
		public int read(ByteBuffer dst);//从通道里面读取数据
		public final SelectionKey register(Selector sel,int ops,Object att);//注册一个选择器并设置监听事件，最后一个参数可以设置共享数据
		public final void close();//关闭通道
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

package com.clover.nio.groupchat;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;

public class GroupChatServer {
    // 定义属性
    private Selector selector;
    private ServerSocketChannel listenChannel;
    private static final int PORT=6667;

    // 构造器
    // 初始化工作
    public GroupChatServer(){
        try {
            // 得到选择器
            selector = Selector.open();
            // 得到ServerSocketChannel
            listenChannel = ServerSocketChannel.open();
            // 绑定端口
            listenChannel.socket().bind(new InetSocketAddress(PORT));
            // 设置非阻塞模式
            listenChannel.configureBlocking(false);
            // 将该listenChannel 注册到Selector
            listenChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e){
            e.printStackTrace();
        }
    }

    // 监听
    public void listen(){
        try {
            // 循环处理
            while(true){
                int count = selector.select();
                if(count > 0){ // 有事件处理
                    // 遍历得到selectionKey集合
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()){
                        // 取出selectionKey
                        SelectionKey key = iterator.next();

                        //根据key对应的通道发生的事件做相应的处理
                        if(key.isAcceptable()){
                            SocketChannel sc = listenChannel.accept();
                            sc.configureBlocking(false);
                            // 将该sc注册到Selector
                            sc.register(selector,SelectionKey.OP_READ);
                            // 输出提示
                            System.out.println(sc.getRemoteAddress() + "上线");
                        }

                        if (key.isReadable()){// 通道发送read事件，即通道是可读状态
                            // 处理读事件(专门写方法)
                            readData(key);
                        }

                        // 手动从集合中移除当前的 SelectionKey ，防止重复操作
                        iterator.remove();
                    }
                }  else {
                    System.out.println("等待......");
                }
            }

        } catch (Exception e){
            e.printStackTrace();
        } finally{

        }
    }

    // 读取客户端数据 通过对应的key获取到对应的通道
    public void readData(SelectionKey key){
        // 定义channel属性
        SocketChannel channel = null;
        try {
            // 得到channel
            channel = (SocketChannel) key.channel();
            // 创建Buffer
            ByteBuffer buffer = ByteBuffer.allocate(1024);//33
            int count = channel.read(buffer);
            // 根据count的值做处理
            if(count > 0){
                // 把缓冲区的数据转成字符串
                String msg = new String(buffer.array());
                // 输出该消息
                System.out.println("from 客户端：" + msg);

                // 向其它的客户端转发消息(去掉自己)，专门写一个方法来处理
                sendInfoToOtherClients(msg,channel);
            }
        } catch (IOException e){
            try {
                System.out.println(channel.getRemoteAddress() + " 离线了....");
                // 取消注册
                key.cancel();
                // 关闭通道
                channel.close();
            } catch (IOException e1){
                e1.printStackTrace();
            }

        }
    }

    // 转发消息给其它客户端(通道)
    public void sendInfoToOtherClients(String msg,SocketChannel self) throws IOException{
        System.out.println("服务器转发消息中..........");
        // 遍历 所有注册到Selector 上的 SocketChannel，并排除自身
        for (SelectionKey key:selector.keys()){
            // 通过key得到对应的 SocketChannel
            Channel targetChannel = key.channel();

            // 排除自己
            if (targetChannel instanceof SocketChannel && targetChannel != self){
                // 转型
                SocketChannel dest = (SocketChannel)targetChannel;
                // 将msg存储到buffer
                ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
                // 将buffer的数据写入通道
                dest.write(buffer);
            }
        }
    }



    public static void main(String[] args) {
        // 创建服务器对象
        GroupChatServer groupChatServer = new GroupChatServer();

        groupChatServer.listen();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

package com.clover.nio.groupchat;

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.Iterator;
import java.util.Scanner;

public class GroupChatClient {
    // 定义相关的属性
    private final String HOST = "127.0.0.1"; // 服务器的IP
    private final int PORT = 6667;
    private Selector selector;
    private SocketChannel socketChannel;
    private String username;

    // 构造器,完成初始化工作
    public GroupChatClient() throws IOException {
        selector = Selector.open();
        // 连接服务器
        socketChannel = SocketChannel.open(new InetSocketAddress(HOST,PORT));
        // 设置非阻塞
        socketChannel.configureBlocking(false);
        // 将channel注册到selector
        socketChannel.register(selector, SelectionKey.OP_READ);
        // 得到username
        username = socketChannel.getLocalAddress().toString();

        System.out.println(username + " is ok.....");
    }

    // 向服务器发送消息
    public void sendInfo(String info){
        info = username + "说：" + info;

        try{
            socketChannel.write(ByteBuffer.wrap(info.getBytes()));
        } catch (IOException e){
            e.printStackTrace();
        }
    }

    // 读取从服务器端回复的消息
    public void readInfo(){
        try {
            int readChannels = selector.select();
            if (readChannels > 0){
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()){
                    SelectionKey key = iterator.next();
                    if(key.isReadable()){
                        SocketChannel channel = (SocketChannel)key.channel();
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        channel.read(buffer);
                        // 将读取到的数据转化为字符串
                        String msg = new String(buffer.array());
                        // 输出消息并去掉头尾空格
                        System.out.println(msg.trim());
                    }
                    // 手动从集合中移除当前的 SelectionKey ，防止重复操作
                    iterator.remove();
                }
            } else {
//                System.out.println("没有可以用的通道....");
            }
        } catch (IOException e){
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception{
        // 启动我们的客户端
        GroupChatClient chatClient = new GroupChatClient();

        // 启动一个线程,每隔3秒，读取从服务器端发送的数据
        new Thread(){
            public void run(){
                while (true){
                    chatClient.readInfo();
                    try {
                        Thread.currentThread().sleep(3000);
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            }
        }.start();

        // 发送数据给服务器端
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNextLine()){
            String s = scanner.nextLine();
            chatClient.sendInfo(s);
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

package com.clover.nio.zerocopy;

import java.io.DataInputStream;
import java.net.ServerSocket;
import java.net.Socket;

public class OldIOServer {

    public static void main(String[] args) throws Exception {
        ServerSocket serverSocket = new ServerSocket(7001);

        while (true) {
            Socket socket = serverSocket.accept();
            DataInputStream dataInputStream = new DataInputStream(socket.getInputStream());

            try {
                byte[] byteArray = new byte[4096];

                while (true) {
                    int readCount = dataInputStream.read(byteArray, 0, byteArray.length);

                    if (-1 == readCount) {
                        break;
                    }
                }
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

package com.clover.nio.zerocopy;

import java.io.DataOutputStream;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.Socket;

public class OldIOClient {

    public static void main(String[] args) throws Exception {
        Socket socket = new Socket("localhost", 7001);

        String fileName = "protoc-3.6.1-win32.zip";
        InputStream inputStream = new FileInputStream(fileName);

        DataOutputStream dataOutputStream = new DataOutputStream(socket.getOutputStream());

        byte[] buffer = new byte[4096];
        long readCount;
        long total = 0;

        long startTime = System.currentTimeMillis();

        while ((readCount = inputStream.read(buffer)) >= 0) {
            total += readCount;
            dataOutputStream.write(buffer);
        }

        System.out.println("发送总字节数： " + total + ", 耗时： " + (System.currentTimeMillis() - startTime));

        dataOutputStream.close();
        socket.close();
        inputStream.close();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

package com.clover.nio.zerocopy;

import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

public class NewIOServer {
  public static void main(String[] args) throws Exception{
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    InetSocketAddress inetSocketAddress = new InetSocketAddress(7001);
    ServerSocket serverSocket = serverSocketChannel.socket();
    serverSocket.bind(inetSocketAddress);

    //创建Buffer
    ByteBuffer buffer = ByteBuffer.allocate(4096);

    while (true){
      SocketChannel socketChannel = serverSocketChannel.accept();

      int readCount = 0;

      while (-1 != readCount){
        try {
          readCount = socketChannel.read(buffer);
        } catch (Exception e){
          e.printStackTrace();
        }

        // 将buffer倒带，因为这个buffer这次使用后下次还需使用
        buffer.rewind();// 使 position = 0 mark=-1 作废
      }
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

package com.clover.nio.zerocopy;

import java.io.FileInputStream;
import java.net.InetSocketAddress;
import java.nio.channels.FileChannel;
import java.nio.channels.SocketChannel;

public class NewIOClient {
  public static void main(String[] args) throws Exception{
      SocketChannel socketChannel = SocketChannel.open();
      socketChannel.connect(new InetSocketAddress("localhost",7001));
      String filename = "protoc-3.6.1-win32.zip";

      // 得到一个文件channel
      FileChannel fileChannel = new FileInputStream(filename).getChannel();

      // 准备发送
      long startTime = System.currentTimeMillis();

      // 在Linux下，一个transferTo 方法就可以完成传输
      // 在windows下，一次调用transferTo只能发送8M文件，就需要分段传输文件，而且要注意传输位置
      long transferCount = fileChannel.transferTo(0, fileChannel.size(), socketChannel);

      System.out.println("发送的总的字节数 =" + transferCount + " 耗时："+(System.currentTimeMillis()-startTime));

      // 关闭通道
      fileChannel.close();
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

package com.clover.netty.simple;

import io.netty.bootstrap.ServerBootstrap;
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;

public class NettyServer {
  public static void main(String[] args) throws Exception{
      /*
       * 说明
       * 1.创建两个线程组 bossGroup 和 workerGroup
       * 2.bossGroup 只是处理连接请求，真正的和客户端业务处理，会交给 workerGroup 完成
       * 3.两个线程组都是无线循环
       */
      EventLoopGroup bossGroup = new NioEventLoopGroup();
      EventLoopGroup workerGroup = new NioEventLoopGroup();

      try {
          // 创建服务器的启动对象，配置参数
          ServerBootstrap bootstrap = new ServerBootstrap();

          // 使用链式编程来进行设置
          bootstrap.group(bossGroup,workerGroup)// 设置两个线程组
                  .channel(NioServerSocketChannel.class)// 使用NioSocketChannel作为服务器的通道实现类型
                  .option(ChannelOption.SO_BACKLOG,128)// 设置线程队列等待连接个数
                  .childOption(ChannelOption.SO_KEEPALIVE,true)// 设置保持活动连接状态
                  .childHandler(new ChannelInitializer<SocketChannel>() { // 创建一个通道初始化对象(匿名对象)
                      // 向workerGroup中EventLoop所关联的通道对应的pipeline设置处理器
                      @Override
                      protected void initChannel(SocketChannel ch) throws Exception {
                          ch.pipeline().addLast(new NettyServerHandler());
                      }
                  });// 给我们的workerGroup的 EventLoop对应的管道设置处理器

          System.out.println("服务器 is ready");

          /*
           * ChannelFuture 在Netty中的所有的I/O操作都是异步执行的，这就意味着任何一个I/O操作会立刻返回，不保证
		   * 在调用结束的时候操作会执行完成。因此，会返回一个ChannelFuture的实例，通过这个实例可以获取当前I/O操
		   * 作的状态。
           */

          // 绑定一个端口，并且同步，生成了一个ChannelFuture 对象
          // 启动服务器(并绑定端口)
          // 为什么使用同步？因为Netty是基于异步操作的，如果不使用同步，可能服务器还未启动就执行下面的语句了，就会产生异常
          ChannelFuture cf = bootstrap.bind(6668).sync();

          // 对关闭通道进行监听(只有当你有一个关闭通道这样的事件发生时，才会去进行处理)
          // 这里为什么也需要使用同步？因为不适用同步，他就会跳过这个语句直接执行finally中的语句关闭线程组了
		  // closeFuture().sync() 不是用于关闭 Channel ，而是允许你阻塞直到 Channel 关闭，然后执行一些额外的操作  
		  // 它有一个监听的作用，当通道关闭的时候，它会返回一个 ChannelFuture,这个里面包含了IO操作的状态
          cf.channel().closeFuture().sync();
      } finally{
          bossGroup.shutdownGracefully();
          workerGroup.shutdownGracefully();
      }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

package com.clover.netty.simple;

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

/*
 * 说明
 * 1.我们自定义一个Handler需要继承netty规定好的某个HandlerAdapter(规范)
 * 2.这时我们自定义一个Handler，才能称之为一个handler
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    /*
     * 读取数据(这里我们可以读取客户端发送的数据)
     * 1.ChannelHandlerContext ctx：上下文对象，含有管道pipeline(业务逻辑处理)，通道channel(数据读写处理)，地址
     * 2.Object msg：就是客户端发送的数据，默认时Object类型
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server ctx =" + ctx);
        // 将msg转化成一个ByteBuf
        // ByteBuf时Netty提供的，不是NIO的ByteBuffer
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("客户端发送的消息是：" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端地址：" + ctx.channel().remoteAddress());
    }

    // 数据读取完毕返回给客户端的消息
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // writeAndFlush 是 write方法 + flush方法
        // 将数据写入到缓冲区，并刷新
        // 一般讲，我们对这个发送的数据需要进行编码
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello,客户端",CharsetUtil.UTF_8));
    }

    // 当出现异常时，一般是需要关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.channel().close();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

package com.clover.netty.simple;

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;

public class NettyClient {
  public static void main(String[] args) throws Exception{
      // 客户端需要一个事件循环组
      EventLoopGroup group = new NioEventLoopGroup();

      try {
          // 创建一个客户端启动对象
          // 注意客户端使用的不是ServerBootstrap，而是Bootstrap
          Bootstrap bootstrap = new Bootstrap();

          // 设置相关参数,链式编程
          bootstrap.group(group) // 设置线程组
                  .channel(NioSocketChannel.class) // 设置客户端通道的实现类型(将来使用反射处理)
                  .handler(new ChannelInitializer<SocketChannel>() {
                      @Override
                      protected void initChannel(SocketChannel ch) throws Exception {
                          ch.pipeline().addLast(new NettyClientHandler());// 加入自己的处理器
                      }
                  });
          System.out.println("客户端 is ok");

          // 启动客户端去连接服务器端
          // 关于 ChannelFuture 后面要分析，涉及到Netty的异步模型
          ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();

          // 给关闭通道进行监听(sync是让其为阻塞到通道关闭后做额外的操作)
          channelFuture.channel().closeFuture().sync();
      }finally{
          group.shutdownGracefully();
      }
  }
}