Java NIO学习与记录（四）： SocketChannel与BIO服务器

发表于 分类于 阅读次数： 本文字数： 3.5k 阅读时长 ≈ 13 分钟

SocketChannel可以创建连接TCP服务的客户端，用于为服务发送数据，SocketChannel的写操作和连接操作在非阻塞模式下不会发生阻塞，这篇文章里的客户端采用SocketChannel实现，利用线程池模拟多个客户端并发访问服务端的情景。服务端仍然采用ServerSocket来实现，主要用来看下阻塞模式下的服务端在并发访问时所做出的的处理。

一、使用SocketChannel实现一个客户端

代码块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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
private static ExecutorService ctp = Executors.newCachedThreadPool();

    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            ctp.submit(IOTest::client); //并发十个客户端连接过去
        }
    }
    
    public static void client() {
        ByteBuffer buffer = ByteBuffer.allocate(1024); //定义缓冲区
        SocketChannel socketChannel = null;
        try {
            socketChannel = SocketChannel.open(); //打开SocketChannel
            socketChannel.configureBlocking(false); //设置为非阻塞模式
            socketChannel.connect(new InetSocketAddress("127.0.0.1", 2333)); //连接服务
            while (true) {
                if(socketChannel.finishConnect()){ //这里的finishConnect是尝试连接，有可能返回false，因此使用死循环进行连接检查，确保连接已经正常建立。
                    System.out.println("客户端已连接到服务器");
                    int i = 0;
                    while (i < 5) {
                        TimeUnit.SECONDS.sleep(1); //隔一秒钟写一条
                        String info = "来自客户端的第" + (i++) + "条消息";
                        buffer.clear();
                        buffer.put(info.getBytes());
                        buffer.flip();
                        while (buffer.hasRemaining()) {
                            socketChannel.write(buffer); //给服务写消息
                        }
                    }
                    break;
                }
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            try {
                if (socketChannel != null) {
                    System.out.println("客户端Channel关闭");
                    socketChannel.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

上面会同时产生10个客户端去连接服务端

二、使用ServerSocket实现一个BIO的TCP服务

代码块2
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
ServerSocket serverSocket = null;
        int recvMsgSize = 0;
        InputStream in = null;
        try {
            serverSocket = new ServerSocket(2333); //开一个监听2333端口的TCP服务
            byte[] recvBuf = new byte[1024];
            while (true) {
                Socket clntSocket = serverSocket.accept(); //探听有没有新的客户端连接进来，没有就阻塞
                SocketAddress clientAddress = clntSocket.getRemoteSocketAddress(); //通过跟服务连接上的客户端socket，拿到客户端地址
                System.out.println("连接成功，处理客户端：" + clientAddress);
                in = clntSocket.getInputStream(); //数据流
                while ((recvMsgSize = in.read(recvBuf)) != -1) { //读取发送的数据，当客户端未断开连接，且不往服务端发数据的时候，说明一直处于准备读的状态，会一直阻塞下去，直到有数据写入（读就绪）
                    byte[] temp = new byte[recvMsgSize];
                    System.arraycopy(recvBuf, 0, temp, 0, recvMsgSize);
                    System.out.println("收到客户端" + clientAddress + "的消息内容：" + new String(temp)); //打印消息
                }
                System.out.println("-----------------------------------");
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (serverSocket != null) {
                    System.out.println("socket关闭！");
                    serverSocket.close();
                }
                if (in != null) {
                    System.out.println("stream连接关闭！");
                    in.close();
                }
            } catch (IOException 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
32
33
34
连接成功，处理客户端：/127.0.0.1:54688
收到客户端/127.0.0.1:54688的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:54688的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:54688的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:54688的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:54688的消息内容：来自客户端的第4条消息
-----------------------------------
连接成功，处理客户端：/127.0.0.1:54680
收到客户端/127.0.0.1:54680的消息内容：来自客户端的第0条消息来自客户端的第1条消息来自客户端的第2条消息来自客户端的第3条消息来自客户端的第4条消息
-----------------------------------
连接成功，处理客户端：/127.0.0.1:54689
收到客户端/127.0.0.1:54689的消息内容：来自客户端的第0条消息来自客户端的第1条消息来自客户端的第2条消息来自客户端的第3条消息来自客户端的第4条消息
-----------------------------------
连接成功，处理客户端：/127.0.0.1:54682
收到客户端/127.0.0.1:54682的消息内容：来自客户端的第0条消息来自客户端的第1条消息来自客户端的第2条消息来自客户端的第3条消息来自客户端的第4条消息
-----------------------------------
连接成功，处理客户端：/127.0.0.1:54683
收到客户端/127.0.0.1:54683的消息内容：来自客户端的第0条消息来自客户端的第1条消息来自客户端的第2条消息来自客户端的第3条消息来自客户端的第4条消息
-----------------------------------
连接成功，处理客户端：/127.0.0.1:54684
收到客户端/127.0.0.1:54684的消息内容：来自客户端的第0条消息来自客户端的第1条消息来自客户端的第2条消息来自客户端的第3条消息来自客户端的第4条消息
-----------------------------------
连接成功，处理客户端：/127.0.0.1:54685
收到客户端/127.0.0.1:54685的消息内容：来自客户端的第0条消息来自客户端的第1条消息来自客户端的第2条消息来自客户端的第3条消息来自客户端的第4条消息
-----------------------------------
连接成功，处理客户端：/127.0.0.1:54681
收到客户端/127.0.0.1:54681的消息内容：来自客户端的第0条消息来自客户端的第1条消息来自客户端的第2条消息来自客户端的第3条消息来自客户端的第4条消息
-----------------------------------
连接成功，处理客户端：/127.0.0.1:54686
收到客户端/127.0.0.1:54686的消息内容：来自客户端的第0条消息来自客户端的第1条消息来自客户端的第2条消息来自客户端的第3条消息来自客户端的第4条消息
-----------------------------------
连接成功，处理客户端：/127.0.0.1:54687
收到客户端/127.0.0.1:54687的消息内容：来自客户端的第0条消息来自客户端的第1条消息来自客户端的第2条消息来自客户端的第3条消息来自客户端的第4条消息
-----------------------------------

可以看到，消息是按照顺序，一个一个连接进来，然后完成处理的，至于后面的消息为什么会被合并成一个，也是这个原因，因为阻塞，所以等第一个连接逐条输出完成后，第二个连接进来，这时很可能客户端的SocketChannel已经将十条消息全部写入channel，等第一个连接处理完成后，接到第二条消息时就已经是全部的消息了，因此一次性输出，后面的合并也是这个原因（主要客户端使用NIO实现，因此写和连接服务不会发生阻塞，因此在第次个请求服务端还在处理时，其余的客户端数据也在执行并写入通道，最终服务端处理完第一个连接，然后继续接收第二个连接时，数据便是完整的5条数据了）。

上面的服务端是一个典型的阻塞IO的服务，accept在没有连接进来时会发生阻塞，read在客户端连接没关闭，且不再写消息时，服务端的read将一直处于读等待状态并阻塞，直到收到新的消息转为读就绪才会继续往下执行（这就是上面例子里第一个进来的连接可以逐条输出的原因），完全串行化，过程如下图：

图1

下面，来改造下服务端，让其处理能力更好一些，除了accept，下面的处理逻辑全部交给线程池处理：

代码块3
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
while (true) {
                Socket clntSocket = serverSocket.accept(); //探听有没有新的客户端连接进来，没有就阻塞
                SocketAddress clientAddress = clntSocket.getRemoteSocketAddress(); //通过跟服务连接上的客户端socket，拿到客户端地址
                System.out.println("连接成功，处理客户端：" + clientAddress);


                ctp.execute(() -> {
                    int recvMsgSize = 0;
                    InputStream in = null; //数据流
                    try {
                        in = clntSocket.getInputStream();
                        while ((recvMsgSize = in.read(recvBuf)) != -1) { //读取发送的数据，当客户端未断开连接，且不往服务端发数据的时候，说明一直处于准备读的状态，会一直阻塞下去，直到有数据写入（读就绪）
                            byte[] temp = new byte[recvMsgSize];
                            System.arraycopy(recvBuf, 0, temp, 0, recvMsgSize);
                            System.out.println("收到客户端" + clientAddress + "的消息内容：" + new String(temp)); //打印消息
                        }
                        System.out.println("-----------------------------------");
                    } catch (IOException 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
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
连接成功，处理客户端：/127.0.0.1:55259
连接成功，处理客户端：/127.0.0.1:55265
连接成功，处理客户端：/127.0.0.1:55266
连接成功，处理客户端：/127.0.0.1:55257
连接成功，处理客户端：/127.0.0.1:55260
连接成功，处理客户端：/127.0.0.1:55258
连接成功，处理客户端：/127.0.0.1:55261
连接成功，处理客户端：/127.0.0.1:55262
连接成功，处理客户端：/127.0.0.1:55263
连接成功，处理客户端：/127.0.0.1:55264
收到客户端/127.0.0.1:55265的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:55266的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:55258的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:55257的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:55261的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:55262的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:55263的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:55260的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:55259的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:55264的消息内容：来自客户端的第0条消息
收到客户端/127.0.0.1:55265的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:55266的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:55258的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:55257的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:55261的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:55260的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:55262的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:55263的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:55259的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:55264的消息内容：来自客户端的第1条消息
收到客户端/127.0.0.1:55266的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:55262的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:55261的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:55260的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:55263的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:55257的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:55265的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:55258的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:55259的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:55264的消息内容：来自客户端的第2条消息
收到客户端/127.0.0.1:55258的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:55266的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:55257的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:55262的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:55261的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:55263的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:55265的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:55260的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:55264的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:55259的消息内容：来自客户端的第3条消息
收到客户端/127.0.0.1:55266的消息内容：来自客户端的第4条消息
收到客户端/127.0.0.1:55265的消息内容：来自客户端的第4条消息
-----------------------------------
收到客户端/127.0.0.1:55263的消息内容：来自客户端的第4条消息
收到客户端/127.0.0.1:55261的消息内容：来自客户端的第4条消息
-----------------------------------
收到客户端/127.0.0.1:55260的消息内容：来自客户端的第4条消息
-----------------------------------
收到客户端/127.0.0.1:55262的消息内容：来自客户端的第4条消息
-----------------------------------
收到客户端/127.0.0.1:55257的消息内容：来自客户端的第4条消息
-----------------------------------
-----------------------------------
收到客户端/127.0.0.1:55258的消息内容：来自客户端的第4条消息
-----------------------------------
-----------------------------------
收到客户端/127.0.0.1:55264的消息内容：来自客户端的第4条消息
-----------------------------------
收到客户端/127.0.0.1:55259的消息内容：来自客户端的第4条消息
-----------------------------------

消息被分开了，接收连接虽然仍然是串行，但实际的处理速度在多线程的帮助下已经比之前快很多了，流程如下图：

图2

三、BIO总结

综合看下来，传统的阻塞IO，按照图2的方式进行，虽然利用多线程避免了read等操作的阻塞对accept的影响，提高了处理效率，但想象下，如果现在存在高并发的情况，图2的模型如果不使用线程池，就会创建大量线程，会发生大量的线程上下文切换，影响整体效率，并且会影响新的线程，如果使用线程池，虽然某种程度上避免了线程的创建和上下文切换的量级，但是在大量并发的场景下，会发生排队，一旦发生排队，紧接着就会影响到accept。