NIO

Posted 2020-09-15 艺海浮台

 1 /**
 2  * Server,NIO本质是非阻塞。 Selector SelectionKey ServerSocketChannel SocketChannel
 3  */
 4 public class MyServerSocketChannel {
 5     public static void main(String[] args) {
 6         Selector sel = null ;
 7         //
 8         try {
 9             sel = Selector.open();                                            //开启挑选器
10             ServerSocketChannel ssc = ServerSocketChannel.open();            //开启服务器SocketChannel
11             InetSocketAddress addr = new InetSocketAddress("0.0.0.0", 8888);
12             ssc.bind(addr);
13             ssc.configureBlocking(false);                                    //配置非阻塞
14             ssc.register(sel, SelectionKey.OP_ACCEPT);                        //***** 在挑选器中注册ssc通道 *****
15         } catch (Exception e) {
16             e.printStackTrace();
17         }
18         
19         SocketChannel sc0 = null;
20         ByteBuffer buf = null;
21         // CharBuffer cbuf = CharBuffer.allocate(1024);
22         while (true) {
23             // 开始挑选
24             try {
25                 sel.select();                                                //开始挑选
26             } catch (Exception e) {
27                 e.printStackTrace();
28             }
29             // 处理发生的事件
30             Set<SelectionKey> keys = sel.selectedKeys();                    //获得selectedKey集合
31             for (SelectionKey key : keys) {
32                 try {
33                     // 是否是accept事件
34                     if (key.isAcceptable()) {
35                         // 得到服务器通道
36                         ServerSocketChannel ssc0 = (ServerSocketChannel) key.channel();//ssc0和ssc是同一个服务器通道，只是不同的引用
37                         // 接受客户端连接，返回SocketChannel
38                         sc0 = ssc0.accept();
39                         System.out.println(getClientInfo(sc0.socket()) + " : 上线了!" );
40                         // 注册非阻塞
41                         sc0.configureBlocking(false);
42                         // 在挑选器中注册新产生的SocketChannel
43                         sc0.register(sel, SelectionKey.OP_READ | SelectionKey.OP_WRITE | SelectionKey.OP_CONNECT);
44                         //
45                     }
46                     // 可以读取数据了。
47                     if (key.isReadable()) {
48                         // 得到SocketChannel
49                         sc0 = (SocketChannel) key.channel();
50                         ByteArrayOutputStream baos = new ByteArrayOutputStream();
51 
52                         //
53                         buf = ByteBuffer.allocate(1024);
54                         // 读取客户端发送的数据
55                         while (sc0.read(buf) != 0) {
56                             buf.flip();
57                             baos.write(buf.array(), 0, buf.limit());
58                             buf.clear();                        //重置buf
59                         }
60                         // 构造缓冲区，写回去client.
61                         String str = "hello : " + new String(baos.toByteArray());
62                         // 输出内容
63                         System.out.println(getClientInfo(sc0.socket()) + str);
64                         
65                         //回传信息
66                         buf = ByteBuffer.allocate(str.getBytes().length);
67                         buf.put(str.getBytes());
68                         buf.flip();
69                         sc0.write(buf);
70                         buf.clear();
71                     }
72                 } catch (Exception e) {
73                     //从selector注销通道
74                     key.cancel();
75                     //
76                     if(key.channel() instanceof SocketChannel){
77                         Socket s = ((SocketChannel)key.channel()).socket();
78                         System.out.println(getClientInfo(s) + " : 下线了！");
79                     }
80                 }
81             }
82             keys.clear();
83         }
84     }
85 
86     /**
87      * 获得客户端信息
88      */
89     private static String getClientInfo(Socket socket) {
90         InetSocketAddress addr = (InetSocketAddress) socket.getRemoteSocketAddress();
91         String ip = addr.getAddress().getHostAddress();
92         String port = addr.getPort() + "";
93         return "[" + ip + ":" + port + "] ";
94     }
95 }

2、客户端

 1 /**
 2  * 客户端
 3  */
 4 public class MyClientSocketChannel {
 5     public static void main(String[] args) throws Exception {
 6         Selector sel = Selector.open();                //挑选器
 7         SocketChannel sc = SocketChannel.open();    //开启通道
 8         InetSocketAddress addr = new InetSocketAddress("localhost", 8888);//服务器地址
 9         sc.connect(addr);                            //连接
10         
11         sc.configureBlocking(false);                //*****非阻塞模式
12         sc.register(sel, SelectionKey.OP_READ);        //注册read事件
13         
14         new Sender(sc).start();                        //开启线程发送消息
15         
16         //
17         ByteBuffer buf = ByteBuffer.allocate(1024);
18         //开始挑选
19         while(true){
20             sel.select();
21             ByteArrayOutputStream baos = new ByteArrayOutputStream();
22             while(sc.read(buf) != 0){
23                 buf.flip();
24                 baos.write(buf.array(),0,buf.limit());
25                 buf.clear();
26             }
27             String str = new String(baos.toByteArray());
28             System.out.println(str);
29         }
30     }
31 }

3、多线程分支读取数据的输入

 1 /**
 2  * 发送器线程
 3  */
 4 public class Sender extends Thread{
 5     private SocketChannel sc ;
 6     public Sender(SocketChannel sc){
 7         this.sc = sc ;
 8     }
 9     public void run() {
10         try {
11             //读取console内容，写入到sc
12             BufferedReader br = new BufferedReader(
13                     new InputStreamReader(System.in));
14             String line = null ;
15             ByteBuffer buf = null ;
16             while((line = br.readLine()) != null){
17                 buf = ByteBuffer.allocate(1024);
18                 buf.put(line.getBytes());
19                 buf.flip();
20                 sc.write(buf);
21                 buf.clear();
22             }
23         } catch (Exception e) {
24             e.printStackTrace();
25         }
26     }
27 }