• Java分别用BIO、NIO实现简单的客户端服务器通信

    分别用BIO、NIO实现客户端服务器通信

    前言：
    Java I/O模型发展以及Netty网络模型的设计思想

    BIO

    Java BIO是Java平台上的BIO（Blocking I/O）模型，是Java中用于实现同步阻塞网络编程的一种方式。 在Java中，使用BIO模型需要通过Socket和ServerSocket类来完成网络连接和数据传输，但是由于BIO是同步阻塞的，所以会导致线程阻塞和资源浪费的问题。
    因此，在高并发的网络编程场景中，通常会选择使用NIO（Non-blocking I/O）模型或者Netty等框架来实现。

    在这里插入图片描述
    服务端类代码

    package com.yu.io.bio;

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

public class BIOServer {
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(8081);
        while (true) {
            System.out.println("wait connect...");
            Socket clientSocket = serverSocket.accept();
            System.out.println(clientSocket.getPort() + "connected");
            System.out.println("start read...");

            byte[] readArr = new byte[1024];
            int read = clientSocket.getInputStream().read(readArr);
            if (read != -1) {
                System.out.println("read info : " + new String(readArr,0,read));
            }
            System.out.println("end read...");
            byte[] resBytes = "server response".getBytes();
            clientSocket.getOutputStream().write(resBytes);
            System.out.println("response info : " + new String(readArr,0,read));
            clientSocket.getOutputStream().flush();
        }
    }
}


    • 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.yu.io.bio;

import java.io.IOException;
import java.net.Socket;

public class BIOClient {
    public static void main(String[] args) throws IOException {
        Socket clientSocket = new Socket("127.0.0.1",8081);
        byte[] resBytes = "client response".getBytes();
        clientSocket.getOutputStream().write(resBytes);
        System.out.println("response info : " + new String(resBytes));
        clientSocket.getOutputStream().flush();
        byte[] readArr = new byte[1024];
        int read = clientSocket.getInputStream().read(readArr);
        if (read != -1) {
            System.out.println("read info : " + new String(readArr,0,read));
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    NIO

    Java NIO 能够支持非阻塞网络编程，可以理解为new io 或者no blok io 我更喜欢称之为new io，因为他不仅仅实现了非阻塞的网络编程方式，同时也封装了常用的网络编程api，更重要的是引入了多路复用器Selector的概念

    在这里插入图片描述

    下面的代码只是展示NIO非阻塞的实现，并没有展示NIO的真正用法

    NIO演示（无Selector）

    服务端类代码

    package com.yu.io.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;

public class NIOServer {

    public static List<SocketChannel> socketChannelList = new ArrayList<>();
    public static void main(String[] args) throws IOException {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(8081));
        serverSocketChannel.configureBlocking(false);
        System.out.println("server start...");
        while (true) {
            SocketChannel socketChannel = serverSocketChannel.accept();
            if (socketChannel != null) {
                System.out.println(socketChannel.socket().getPort() + " connected");
                socketChannel.configureBlocking(false);
                socketChannelList.add(socketChannel);
            }
            List<SocketChannel> rmChannelList = new ArrayList<>();
            for (SocketChannel channel : socketChannelList) {
                try {
                    doChannel(rmChannelList, channel);
                } catch (IOException ioException) {
                    //有客户端断开连接
                    System.out.println(channel.socket().getPort() + " disconnected");
                    channel.close();
                    rmChannelList.add(channel);
                }
            }
            socketChannelList.removeAll(rmChannelList);
        }
    }

    private static void doChannel(List<SocketChannel> rmChannelList, SocketChannel channel) throws IOException {
        ByteBuffer readByteBuffer = ByteBuffer.allocate(2048);
        int read = channel.read(readByteBuffer);
        String readStr = new String(readByteBuffer.array());
        if (read > 0) {
            System.out.println(channel.socket().getPort() + " : " + readStr);
            if (readStr.contains("hello")) {
                ByteBuffer sendByteBuffer = ByteBuffer.wrap("hello! I am robot.".getBytes());
                channel.write(sendByteBuffer);
                System.out.println("me : " + new String(sendByteBuffer.array()));
            }
            if (readStr.contains("old")) {
                ByteBuffer sendByteBuffer = ByteBuffer.wrap("I am 1 years old.".getBytes());
                channel.write(sendByteBuffer);
                System.out.println("me : " + new String(sendByteBuffer.array()));
            }
            if (readStr.contains("bey")) {
                ByteBuffer sendByteBuffer = ByteBuffer.wrap("see you.".getBytes());
                channel.write(sendByteBuffer);
                System.out.println("me : " + new String(sendByteBuffer.array()));
            }
        }
        if (read == -1) {
            rmChannelList.add(channel);
        }
    }
}

    • 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

    客户端类代码

    package com.yu.io.nio;

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

public class NIOClient {
    public static void main(String[] args) throws IOException {
        SocketChannel clientSocketChannel = SocketChannel.open();
        clientSocketChannel.connect(new InetSocketAddress("127.0.0.1",8081));
        clientSocketChannel.configureBlocking(false);
        String sendStr = "hello! I am client " + clientSocketChannel.socket().getPort() + ".";
        ByteBuffer sendByteBuffer = ByteBuffer.wrap(sendStr.getBytes());
        clientSocketChannel.write(sendByteBuffer);
        System.out.println("me : " + new String(sendByteBuffer.array()));
        int msgSize = 0;
        while (msgSize < 10) {
            ByteBuffer readByteBuffer = ByteBuffer.allocate(1024);
            int read = clientSocketChannel.read(readByteBuffer);
            String readStr = new String(readByteBuffer.array());
            if (read > 0) {
                System.out.println("robot : " + readStr);
                msgSize ++;
                ByteBuffer resByteBuffer = null;
                if (readStr.contains("hello")) {
                    resByteBuffer = ByteBuffer.wrap("how old are you?.".getBytes());
                    clientSocketChannel.write(resByteBuffer);
                    System.out.println("me : " + new String(resByteBuffer.array()));
                    resByteBuffer.clear();
                }
                if (readStr.contains("old")) {
                    resByteBuffer = ByteBuffer.wrap("en, place say hello!".getBytes());
                    clientSocketChannel.write(resByteBuffer);
                    System.out.println("me : " + new String(resByteBuffer.array()));
                    resByteBuffer.clear();
                }
            }
        }
        ByteBuffer resByteBuffer = ByteBuffer.wrap("bey bey!".getBytes());
        clientSocketChannel.write(resByteBuffer);
        System.out.println("me : " + new String(resByteBuffer.array()));
        resByteBuffer.clear();
        clientSocketChannel.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

    NIO演示（Selector）

    无Selector的NIO演示中，显然会出现空转的情况，以及无效连接的处理问题，这些问题都会影响性能。
    NIO提供Selector多路复用器，优化上述问题
    以下demo实现服务器与客户端通信，相互发消息。并由服务器转发给其他客户端（广播功能）

    服务端类代码

    server main类

    import java.io.IOException;

public class NIOServerMain {
    public static void main(String[] args) throws IOException {
        NIOSelectorServer server = new NIOSelectorServer();
        server.start();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    server run类

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

public class NIOSelectorServer {

    public void start() throws IOException {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(8081));
        serverSocketChannel.configureBlocking(false);
        //注册到selector多路复用器中
        Selector selector = Selector.open();
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("server start...");
        run(selector);
    }

    /**
     * 遍历多路复用器的事件，处理事件
     */
    private void run(Selector selector) throws IOException {
        while (true) {
            //阻塞等待事件
            selector.select();

            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey selectedKey = iterator.next();
                try {
                    if (selectedKey.isAcceptable()) {
                        doAccept(selector, selectedKey);
                    }
                    if (selectedKey.isReadable()) {
                        doReadChannel(selectedKey, selector);
                    }
                } catch (IOException ioException) {
                    //有客户端断开连接
                    disConnect(selectedKey, "exception");
                }
                iterator.remove();
            }
        }
    }

    /**
     * 处理连接事件
     */
    private void doAccept(Selector selector, SelectionKey selectedKey) throws IOException {
        ServerSocketChannel serverChannel = (ServerSocketChannel)selectedKey.channel();
        SocketChannel socketChannel = serverChannel.accept();
        socketChannel.configureBlocking(false);
        socketChannel.register(selector, SelectionKey.OP_READ);
        System.out.println(socketChannel.socket().getPort() + " connected");
    }

    /**
     * 处理接收信息事件
     */
    private void doReadChannel(SelectionKey selectedKey, Selector selector) throws IOException {
        SocketChannel channel = (SocketChannel)selectedKey.channel();
        ByteBuffer readByteBuffer = ByteBuffer.allocate(2048);
        int read = channel.read(readByteBuffer);
        String readStr = "";
        readByteBuffer.flip();
        readStr += StandardCharsets.UTF_8.decode(readByteBuffer);
        if (read > 0) {
            System.out.println(channel.socket().getPort() + " : " + readStr.length() + " - "+ readStr);
            //转发消息（其他客户端）
            broadcast(selectedKey, selector, readStr);
            if (readStr.contains("hello")) {
                sendMsg(channel, "hello! I am robot.");
            }
            if (readStr.contains("old")) {
                sendMsg(channel, "I am 1 years old.");
            }
            if (readStr.contains("bye")) {
                sendMsg(channel, "see you.");
            }
        }
        if (read == -1) {
            //有客户端断开连接
            disConnect(selectedKey, "read = -1");
        }
    }

    /**
     * 连接异常的处理
     */
    private void disConnect(SelectionKey selectedKey, String type) throws IOException {
        SocketChannel channel = (SocketChannel)selectedKey.channel();
        System.out.println(channel.socket().getPort() + " disconnected. " + type);
        selectedKey.cancel();
        channel.close();
    }

    /**
     * 发送消息
     */
    private void sendMsg(SocketChannel channel, String s) throws IOException {
        ByteBuffer sendByteBuffer = ByteBuffer.wrap(s.getBytes());
        channel.write(sendByteBuffer);
        System.out.println("me : " + new String(sendByteBuffer.array()));
    }

    /**
     * 广播
     * 转发消息（给其他客户端）
     */
    private void broadcast(SelectionKey  fromSelectedKey, Selector selector, String readStr) throws IOException {
        Iterator<SelectionKey> selectionKeyIterator = selector.keys().iterator();
        while (selectionKeyIterator.hasNext()) {
            SelectionKey otherKey = selectionKeyIterator.next();
            if (otherKey == fromSelectedKey) {
                continue;
            }
            if (!(otherKey.channel() instanceof SocketChannel)) {
                continue;
            }
            SocketChannel otherChannel = (SocketChannel)otherKey.channel();
            sendMsg(otherChannel, "(转发自 "+ ((SocketChannel)fromSelectedKey.channel()).socket().getPort() + ")" + readStr);
        }
    }
}

    • 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

    客户端代码

    一共构造了两个客户端（消息客户端和looker客户端）， looker客户端优先启动，随后启动消息客户端，消息客户端与服务器的通信会被转发给looker客户端

    look client main类

    import java.io.IOException;

public class NIOClientLookMain {
    public static void main(String[] args) throws IOException, InterruptedException {
        NIOSelectorClient client = new NIOSelectorLookClient();
        client.start(8081, 100);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    msg client main类

    import java.io.IOException;

public class NIOClientMain {
    public static void main(String[] args) throws IOException, InterruptedException {
        NIOSelectorClient lookClient = new NIOSelectorClient();
        lookClient.start(8081, 10);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    client run类

    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.nio.charset.StandardCharsets;
import java.util.Iterator;

public class NIOSelectorClient {
    protected int port;
    protected int size;
    public void start(int port, int size) throws IOException, InterruptedException {
        this.port = port;
        this.size = size;
        SocketChannel clientSocketChannel = SocketChannel.open();
        clientSocketChannel.connect(new InetSocketAddress("127.0.0.1",port));
        clientSocketChannel.configureBlocking(false);
        Selector selector = Selector.open();
        clientSocketChannel.register(selector, SelectionKey.OP_READ);
        //发送开始数据
        sendMsg(clientSocketChannel, "hello! I am client " + clientSocketChannel.socket().getLocalPort() + ".");
        run(selector);
        sendMsg(clientSocketChannel, "bye bye!");
        clientSocketChannel.close();
    }

    /**
     * 遍历多路复用器的事件，处理事件
     */
    protected void run(Selector selector) throws IOException, InterruptedException {
        int msgSize = 0;
        while (msgSize < size) {
            int length=selector.select();
            if(length ==0){
                continue;
            }
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while(iterator.hasNext()){
                SelectionKey selectionKey = iterator.next();
                if(selectionKey.isReadable()){
                    boolean readChannel = false;
                    try {
                        readChannel = doReadChannel(selectionKey, selector);
                    } catch (IOException e) {
                        selectionKey.cancel();
                        System.out.println("robot disconnect, restart connect...");
                        while (true){
                            try {
                                reConnect();
                                return;
                            } catch (IOException ioException) {
                                System.out.println("restart connecting（5s） ");
                                //ioException.printStackTrace();
                                Thread.sleep(5000);
                            }
                        }
                    }
                    if (readChannel) {
                        msgSize ++;
                    }
                }
                iterator.remove();
            }
        }
    }

    protected boolean doReadChannel(SelectionKey selectedKey, Selector selector) throws IOException {
        SocketChannel channel = (SocketChannel)selectedKey.channel();
        ByteBuffer readByteBuffer = ByteBuffer.allocate(2048);
        int read = channel.read(readByteBuffer);
        String readStr = "";
        readByteBuffer.flip();
        readStr += StandardCharsets.UTF_8.decode(readByteBuffer);
        if (read > 0) {
            System.out.println("robot : " + readStr);
            if (readStr.contains("hello")) {
                sendMsg(channel, "how old are you?.");
            }
            if (readStr.contains("old")) {
                sendMsg(channel, "en, place say hello!");
            }
            return true;
        }
        return false;
    }

    protected void sendMsg(SocketChannel channel, String sendStr) throws IOException {
        ByteBuffer sendByteBuffer = ByteBuffer.wrap(sendStr.getBytes());
        channel.write(sendByteBuffer);
        System.out.println("me : " + new String(sendByteBuffer.array()));
    }

    protected void reConnect() throws IOException, InterruptedException {
        NIOSelectorClient client = new NIOSelectorClient();
        client.start(port, size);
    }
}

    • 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

    look client run类

    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.nio.charset.StandardCharsets;

public class NIOSelectorLookClient extends NIOSelectorClient{
    @Override
    public void start(int port, int size) throws IOException, InterruptedException {
        this.port = port;
        this.size = size;
        SocketChannel clientSocketChannel = SocketChannel.open();
        clientSocketChannel.connect(new InetSocketAddress("127.0.0.1",port));
        clientSocketChannel.configureBlocking(false);
        Selector selector = Selector.open();
        clientSocketChannel.register(selector, SelectionKey.OP_READ);
        //发送开始数据
        sendMsg(clientSocketChannel, "I am looker. " + clientSocketChannel.socket().getLocalPort() + ".");
        run(selector);
    }

    @Override
    protected boolean doReadChannel(SelectionKey selectedKey, Selector selector) throws IOException {
        SocketChannel channel = (SocketChannel)selectedKey.channel();
        ByteBuffer readByteBuffer = ByteBuffer.allocate(2048);
        int read = channel.read(readByteBuffer);
        String readStr = "";
        readByteBuffer.flip();
        readStr += StandardCharsets.UTF_8.decode(readByteBuffer);
        if (read > 0) {
            System.out.println("robot : " + readStr.length() + " - "+ readStr);
            if (readStr.contains("bye")) {
                sendMsg(channel, "bye.");
            }
            return true;
        }
        return false;
    }

    @Override
    protected void reConnect() throws IOException, InterruptedException {
        NIOSelectorClient client = new NIOSelectorLookClient();
        client.start(port, size);
    }
}

    • 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
  • 相关阅读:
    代码随想录第34天: 贪心part03
    7 Redis的PipeLine
    回放线上流量利器-GoReplay
    R语言使用colSums函数和is.na函数统计dataframe数据中每个数据列中包含的缺失值的个数
    Mybatis如何使用分页插件pageHelper
    云原生Kubernetes：pod进阶之资源管理与探针
    制造企业如何开展客户满意度调查？民安智库为你解答
    Stability AI推出Stable Audio；ChatGPT：推荐系统的颠覆者
    数据结构 第七章（查找算法）
    天干地支对应的五行
  • 原文地址：https://blog.csdn.net/Ureliable/article/details/132649495