package com.shengsiyuan.nio; import java.io.IOException; import java.net.InetSocketAddress; import java.net.ServerSocket; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.nio.charset.Charset; import java.util.HashMap; import java.util.Map; import java.util.Set; import java.util.UUID; public class NioServer { private static Map<String, SocketChannel> clientMap = new HashMap(); public static void main(String[] args) throws IOException { ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();//创建一个ServerSocketChannel serverSocketChannel.configureBlocking(false);//非阻塞 ServerSocket serverSocket = serverSocketChannel.socket();//通过服务端的channel获取服务端的socket, serverSocket.bind(new InetSocketAddress(8899)); Selector selector = Selector.open();//服务端channel注册到选择器上， //选择器可以关联多个channel对象，这里只有一个channel。 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);//现在关注服务端channel的连接事件， while (true) { try {//异常try()catch{} selector.select();//阻塞，等着关注的事件发生，返回发生的关注事件的数量。 Set<SelectionKey> selectionKeys = selector.selectedKeys();//一个个的事件，现在只有一个OP_ACCEPT事件， selectionKeys.forEach(selectionKey -> { final SocketChannel client; try { //OP_ACCEPT事件 if (selectionKey.isAcceptable()) { //这个selectionKey关联的channel是服务端channel，OP_ACCEPT是服务端channel关注的事件（并注册在选择器上） //所以这里可以强制转为服务端channel， ServerSocketChannel server = (ServerSocketChannel) selectionKey.channel(); client = server.accept();//接受连接，真正接收连接之后返回SocketChannel对象，就是与客户端通信的socket， //对于当前通道，服务端socket就用不上了，用SocketChannel client.configureBlocking(false);//非阻塞的 //转而把SocketChannel注册到选择器，并关注读事件， client.register(selector, SelectionKey.OP_READ); //此时这个选择器有2个socket，一个服务端channel一个SocketChannel，一个关注连接一个关注数据读取， String key = "【" + UUID.randomUUID().toString() + "】"; clientMap.put(key, client); //OP_READ事件 } else if (selectionKey.isReadable()) { client = (SocketChannel) selectionKey.channel();//肯定是客户端关联的socket ByteBuffer readBuffer = ByteBuffer.allocate(1024); int count = client.read(readBuffer);//读到buffer if (count > 0) { readBuffer.flip(); Charset charset = Charset.forName("utf-8");//编码 //发过来的数据转成string String receivedMessage = String.valueOf(charset.decode(readBuffer).array()); //打印客户端发过来的数据 System.out.println(client + ": " + receivedMessage); String senderKey = null; for (Map.Entry<String, SocketChannel> entry : clientMap.entrySet()) { if (client == entry.getValue()) { senderKey = entry.getKey(); break; } } for (Map.Entry<String, SocketChannel> entry : clientMap.entrySet()) { SocketChannel value = entry.getValue(); //先写到ByteBuffer ByteBuffer writeBuffer = ByteBuffer.allocate(1024); writeBuffer.put((senderKey + ": " + receivedMessage).getBytes()); writeBuffer.flip(); //然后buteBuffer写出去到channel value.write(writeBuffer); } } } } catch (Exception ex) { ex.printStackTrace(); } }); selectionKeys.clear();//事件处理完成之后要清空，否则下次还要处理，就会报空指针。就是iter.remove(); } catch (Exception ex) { ex.printStackTrace(); } } } }
package com.shengsiyuan.nio; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; 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.time.LocalDateTime; import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class NioClient { public static void main(String[] args) throws IOException { try { SocketChannel socketChannel = SocketChannel.open();//@587 socketChannel.configureBlocking(false); Selector selector = Selector.open(); //OP_ACCEPT是接受连接，OP_CONNECT是发起连接， socketChannel.register(selector, SelectionKey.OP_CONNECT); socketChannel.connect(new InetSocketAddress("127.0.0.1", 8899)); while (true) { selector.select();//阻塞 Set<SelectionKey> keySet = selector.selectedKeys(); for (SelectionKey selectionKey : keySet) { //已经建立好了连接 if (selectionKey.isConnectable()) { SocketChannel client = (SocketChannel) selectionKey.channel();//就是之前的那个socket @587 //连接是否处于进行状态 if (client.isConnectionPending()) { client.finishConnect();//完成连接，现在连接真正建立好了， //向服务器发送连接建立好了的消息 ByteBuffer writeBuffer = ByteBuffer.allocate(1024); writeBuffer.put((LocalDateTime.now() + " 连接成功").getBytes()); //翻转 writeBuffer.flip(); //写到channel client.write(writeBuffer); ExecutorService executorService = Executors.newSingleThreadExecutor( Executors.defaultThreadFactory());//只有一个线程的线程池 executorService.submit(() -> { while (true) { try { writeBuffer.clear(); InputStreamReader input = new InputStreamReader(System.in); BufferedReader br = new BufferedReader(input); String sendMessage = br.readLine(); //写入到buffer writeBuffer.put(sendMessage.getBytes()); //翻转 writeBuffer.flip(); //buffer写出去 client.write(writeBuffer); } catch (Exception ex) { ex.printStackTrace(); } } }); } //给这个channel注册读取事件， client.register(selector, SelectionKey.OP_READ);//@587 } else if (selectionKey.isReadable()) { SocketChannel client = (SocketChannel) selectionKey.channel();//@587 ByteBuffer readBuffer = ByteBuffer.allocate(1024); int count = client.read(readBuffer); if (count > 0) { String receivedMessage = new String(readBuffer.array(), 0, count);//字节数组转字符串 System.out.println(receivedMessage); } } } keySet.clear();//清除事件，就是清除SelectionKey集合。 } } catch (Exception ex) { ex.printStackTrace(); } } }