Java网络编程入门(四)

￡神魔★判官ぃ 2022-12-11 13:51 169阅读 0赞

前言：这是本人在慕课网上学习 一站式学习Java网络编程 全面理解BIO/NIO/AIO 时所做的笔记，供本人复习之用，在入门二和三中讲述了如何实现一个BIO聊天室和NIO的基本概念后，本文主要讲述如何实现一个基于NIO的聊天室，不保证一定准确。

**目录**

第一章 NIO编程模型

第二章 ChatServer

第三章 ChatClient

3.1 客户端

3.2 处理读入的类

--------------------

# 第一章 NIO编程模型 #

我们要在单个线程中处理多个IO操作，首先我们有一个Selector，我们在服务器端启动我们的ServerSocketChannel，然后将ServerSocketChannel以accept状态注册到Selector中（其实就是只监听ServerSocketChannel的accept事件），当有一个客户端发送连接请求，当ServerSocketChannel接受了这个客户端的连接请求时候，这时候就相当于触发了accept事件，我们去检查selector状态时就会发现这个事件，我们在服务端得到客户端的SocketChannel，并且我们要为新连接的SocketChannel再在Selector注册一个新的事件叫做read，当客户端有信息发过来时，我们调用selector检查时，就会发现客户端发来的了数据，就可以获取数据并处理了。

值得注意的是，上面的操作都是在一个线程中完成的，BIO的话accept是在主线程，但是处理客户端发来的数据是在另一个线程中。同时selector的监听是阻塞的，直到selector监听的channel中有channel处于我们需要的状态，selector才会返回。

假如有第二个客户想要连接，我们继续处理accept事件，并且将客户端channel以read状态注册到selector上。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3E2MTAzNzY2ODE_size_16_color_FFFFFF_t_70][]

# 第二章 ChatServer #

不多分析了，有问题看代码吧，其中有个put报错了，大体思路没问题，正在调。

/**
     * @author ZhangChen
     **/
    public class ChatServer {
        private static final int DEFAULT_PORT = 8888;
        private final String QUIT = "quit";
        private final int BUFFER = 1024;
        private ServerSocketChannel server;
        private Selector selector;
        private ByteBuffer rBuffer = ByteBuffer.allocate(BUFFER);
        private ByteBuffer wBuffer = ByteBuffer.allocate(BUFFER);
        private Charset charset = Charset.forName("UTF-8");
        private int port;
    
        public ChatServer(int port){
            this.port = port;
        }
    
    
        public ChatServer(){
            this(DEFAULT_PORT);
        }
    
        public  void forwardMessage(SocketChannel client,String fwdMsg) throws IOException {
            for (SelectionKey key:selector.keys()){
                if(key.isValid()&&key.channel() instanceof SocketChannel){
                    SocketChannel connectedClient = (SocketChannel) key.channel();
                    if(!client.equals(connectedClient)){
                        wBuffer.clear();
                        //这个put会报错，正在调，大体思路没问题
    wBuffer.put(Byte.parseByte(String.valueOf(charset.encode(getClientName(client)+":"+fwdMsg))));
                        wBuffer.flip();
                        while(wBuffer.hasRemaining()){
                            connectedClient.write(wBuffer);
                        }
                    }
    
                }
            }
        }
    
        private void handles(SelectionKey key) throws IOException {
            if(key.isAcceptable()){
                //Accept事件
                ServerSocketChannel server = (ServerSocketChannel) key.channel();
                SocketChannel client = server.accept();
                client.configureBlocking(false);
                client.register(selector,SelectionKey.OP_READ);
                System.out.println(getClientName(client)+"已连接");
            }
            if(key.isReadable()){
                //read事件 客户端发送消息给服务端
                SocketChannel client = (SocketChannel) key.channel();
                String fwdMsg = receive(client);
                if(fwdMsg.isEmpty()){
                    //客户端异常
                    key.cancel();
                    //修改key，更新selector状态
                    selector.wakeup();
                }else{
                    System.out.println(getClientName(client)+":"+fwdMsg);
                    forwardMessage(client,fwdMsg);
                    //检查用户是否退出
                    if(readyToQuit(fwdMsg)){
                        key.cancel();;
                        selector.wakeup();
                        System.out.println(getClientName(client)+"已断开连接");
                    }
                }
            }
    
        }
        private String receive(SocketChannel client) throws IOException {
            rBuffer.clear();
            while(client.read(rBuffer)>0) {
            }
            rBuffer.flip();
            return String.valueOf(charset.decode(rBuffer));
        }
    
        private String getClientName(SocketChannel client){
            return "客户端["+client.socket().getPort()+"]";
        }
        public boolean readyToQuit(String msg){
            return QUIT.equals(msg);
        }
    
        public synchronized void close(Closeable closeable){
            if(closeable!=null){
                try {
                    closeable.close();
                    System.out.println("关闭serverSocket");
                }catch (IOException e){
                    e.printStackTrace();
                }
            }
        }
    
        public void start(){
            //绑定监听端口
            try {
                server = ServerSocketChannel.open();
                //确保server非阻塞
                server.configureBlocking(false);
                server.socket().bind(new InetSocketAddress(port));
                selector = Selector.open();
                server.register(selector, SelectionKey.OP_ACCEPT);
                System.out.println("启动服务器，监听端口： "+port+"...");
    
                while (true){
                    //等待客户端连接
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    for(SelectionKey key:selectionKeys){
                        //处理被触发的事件
                        handles(key);
                    }
                    //不清空的话会保留上一轮的事件
                    selectionKeys.clear();
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                //当我们关系selector时，selector会把上面注册的channel都解除注册并关闭
                close(selector);
            }
        }
    
        public static void main(String[] args) {
            ChatServer chatServer = new ChatServer(7777);
            chatServer.start();
    
        }
    }

# 第三章 ChatClient #

## 3.1 客户端 ##

/**
     * @author ZhangChen
     **/
    public class ChatClient {
    
        private static final int DEFAULT_SERVER_PORT = 8888;
        private static final String DEFAULT_SERVER_HOST = "127.0.0.1";
        private final String QUIT = "quit";
        private int port;
        private String host;
        private final int BUFFER = 1024;
        private SocketChannel client;
        private Selector selector;
        private ByteBuffer rBuffer = ByteBuffer.allocate(BUFFER);
        private ByteBuffer wBuffer = ByteBuffer.allocate(BUFFER);
        private Charset charset = Charset.forName("UTF-8");
    
        public ChatClient(){
            this(DEFAULT_SERVER_HOST,DEFAULT_SERVER_PORT);
        }
        public ChatClient(String host,int port){
            this.port = port;
            this.host = host;
        }
        //发送消息给服务端
        public void send(String msg) throws IOException {
            if(msg.isEmpty()){
                return;
            }else{
                wBuffer.clear();
                wBuffer.put(charset.encode(msg));
                wBuffer.flip();
                while(wBuffer.hasRemaining()){
                    client.write(wBuffer);
                }
                //检查用户退出
                if(readyToQuit(msg)){
                    close(selector);
                }
            }
        }
        //从服务端接收消息
        public String receive(SocketChannel client) throws IOException {
            rBuffer.clear();
            while(client.read(rBuffer)>0){
    
            }
            rBuffer.flip();
            return String.valueOf(charset.decode(rBuffer));
        }
    
        public boolean readyToQuit(String msg){
            return QUIT.equals(msg);
        }
        public void start (){
            try {
                //创建socket
                client = SocketChannel.open();
                client.configureBlocking(false);
                selector = Selector.open();
                client.register(selector, SelectionKey.OP_CONNECT);
                client.connect(new InetSocketAddress(host,port));
                while(true){
                    //阻塞的
                    selector.select();
                    for(SelectionKey key:selector.keys()){
                        handlers(key);
                    }
                }
            }  catch (IOException e) {
                e.printStackTrace();
            } catch (ClosedSelectorException e){
    
            }finally {
                close(client);
            }
        }
        private void handlers(SelectionKey key) throws IOException {
            //CONNECT事件
            if(key.isConnectable()){
                SocketChannel client = (SocketChannel) key.channel();
                if(client.isConnectionPending()){
                    //连接已建立
                    client.finishConnect();
                    //额外线程处理用户输入信息
                    new Thread(new UserInputHandler(this)).start();
                }
                client.register(selector,SelectionKey.OP_READ);
            }
            else if(key.isReadable()){
                //READ事件
                SocketChannel client = (SocketChannel) key.channel();
                String msg = receive(client);
                if(msg.isEmpty()){
                    //服务器异常
                    close(selector);
                }else{
                    System.out.println(msg);
                }
            }
    
        }
        public  void close(Closeable closeable){
            if(closeable!=null){
                try {
                    closeable.close();
                    System.out.println("关闭serverSocket");
                }catch (IOException e){
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) {
            ChatClient chatClient = new ChatClient();
            chatClient.start();
        }
    
        public void close() {
            close(selector);
        }
    }

## 3.2 处理读入的类 ##

/**
     * @author ZhangChen
     **/
    public class UserInputHandler implements Runnable{
        private ChatClient chatClient;
        public UserInputHandler(ChatClient chatClient){
            this.chatClient = chatClient;
        }
    
        @Override
        public void run() {
            //等待用户输入消息
            BufferedReader reader =null;
            try {
                reader = new BufferedReader(new InputStreamReader(System.in));
                String msg = null;
                while(true){
                    msg = reader.readLine();
                    chatClient.send(msg);
                    if(chatClient.readyToQuit(msg)){
                        chatClient.close();
                        break;
                    }
                }
            }catch (IOException e){
                e.printStackTrace();
            } finally {
                if(reader!=null){
                    try {
                        reader.close();
                    } catch (IOException e1) {
                        e1.printStackTrace();
                    }
                }
            }
    
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3E2MTAzNzY2ODE_size_16_color_FFFFFF_t_70]: /images/20221123/f909fb9d0faa46f0b8a0734d4351b341.png