nio 代码实现简易多人聊天

  1 import java.io.IOException;
  2 import java.net.InetSocketAddress;
  3 import java.nio.ByteBuffer;
  4 import java.nio.channels.*;
  5 import java.nio.charset.Charset;
  6 import java.util.*;
  7 
  8 /**
  9  * @ClassName CharRoomServer
 10  * @Description TODO
 11  * @Author hufeng8
 12  * @Date 2018/8/3 14:39
 13  * @Version 1.0
 14  */
 15 public class CharRoomServer implements Runnable{
 16 
 17     private ServerSocketChannel serverSocketChannel = null;
 18 
 19     private Selector selector = null;
 20 
 21     public static final int PORT_NUM = 1198;
 22 
 23     private boolean active = true;
 24 
 25     private Charset charset = Charset.forName("UTF-8");
 26 
 27     private List<String> users  = new ArrayList<String>();
 28 
 29     private ByteBuffer byteBuffer = ByteBuffer.allocate(2*1024);
 30 
 31     public static final String protocol = "#user#";
 32 
 33     public CharRoomServer() {
 34         this.init();
 35     }
 36 
 37     public void init() {
 38         try {
 39             selector = Selector.open();
 40             serverSocketChannel = ServerSocketChannel.open();
 41             serverSocketChannel.socket().bind(new InetSocketAddress(PORT_NUM));
 42             serverSocketChannel.configureBlocking(false);
 43             serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
 44         } catch (IOException e) {
 45             e.printStackTrace();
 46         }
 47     }
 48 
 49     public void run() {
 50         System.out.println("开始监听。。。。");
 51         while (active) {
 52             try {
 53                 //非阻塞接受连接
 54 //                int s = selector.selectNow();
 55 
 56                 //阻塞连接
 57                 int s = selector.select();
 58 
 59                 System.out.println("服务端接受到连接总数"+selector.keys().size());
 60             } catch (IOException e) {
 61                 e.printStackTrace();
 62             }
 63             System.out.println("服务端接受到的选择连接数"+selector.selectedKeys().size());
 64             Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
 65             while (keys.hasNext()) {
 66                 SelectionKey k = keys.next();
 67                 keys.remove();
 68                 //处理逻辑
 69                 doHandler(selector, k);
 70             }
 71         }
 72     }
 73 
 74     private void doHandler(Selector selector, SelectionKey k) {
 75         try {
 76             //连接事件
 77             if (k.isConnectable()) {
 78                 System.out.println("Connectable 连接事件");
 79             } else if (k.isAcceptable()) {
 80                 ServerSocketChannel ser = (ServerSocketChannel) k.channel();
 81                 if (ser == serverSocketChannel) {
 82                     System.out.println("同一个连接");
 83                 }
 84                 SocketChannel socketChannel = ser.accept();
 85                 socketChannel.configureBlocking(false);
 86                 socketChannel.register(selector, SelectionKey.OP_READ);
 87                 socketChannel.write(charset.encode("please enter login name:"));
 88 
 89                 //设置k为接受事件，准备接受其它请求？
 90 
 91             } else if (k.isReadable()) {
 92                 //获取客户端连接
 93                 SocketChannel socketChannel = (SocketChannel) k.channel();
 94                 StringBuffer content = new StringBuffer();
 95                 int sum = 0;
 96                 try {
 97                     byteBuffer.clear();
 98                     while ((sum = socketChannel.read(byteBuffer)) > 0) {
 99                         byteBuffer.flip();
100                         content.append(charset.decode(byteBuffer));
101                     }
102                     System.out.println(sum);
103                     //判断客户端连接关闭
104                     if (sum == -1) {
105                         socketChannel.close();
106                         System.out.println("1--关闭连接");
107                     }
108 
109                     System.out.println("服务端：监听："+ content.toString());
110                 } catch (Exception e) {
111                     System.out.println("2--关闭连接");
112                     k.cancel();
113                     if (null != socketChannel) {
114                         socketChannel.close();
115                     }
116                 }
117                 if (content.length() > 0) {
118                     //按照协议切分内容
119                     String[] contents = content.toString().split(protocol);
120                     //登陆用户
121                     if (contents != null && contents.length == 1) {
122                         String user = contents[0];
123                         if (users.contains(user)) {
124                             socketChannel.write(charset.encode("登陆用户已存在！"));
125                         } else {
126                             users.add(user);
127                             //获取在线人数
128                             int i = onlineCount(selector);
129                             //广播登陆消息给当前房间所有人
130                             brokerMessage(selector, k, "欢迎"+user+"登陆，当前第"+i+"号");
131                         }
132                     } else if (contents != null && contents.length > 1) {
133                         String message = contents[0] + "say :" + contents[1];
134                         brokerMessage(selector, k, message);
135                     }
136                 }
137             } else if (k.isWritable()) {
138 
139             }
140         } catch (Exception e) {
141             e.printStackTrace();
142         }
143     }
144 
145     /**
146      * 广播消息
147      * @param content
148      */
149     private void brokerMessage(Selector selector, SelectionKey k, String content) {
150         for (SelectionKey key : selector.keys()) {
151             if (key.channel() instanceof SocketChannel && key != k) {
152                 try {
153                     SocketChannel sc = (SocketChannel) key.channel();
154                     sc.write(charset.encode(content));
155                 } catch (IOException e1) {
156                     e1.printStackTrace();
157                 }
158             }
159         }
160     }
161 
162     /**
163      * 统计在线人数
164      * @param selector
165      * @return
166      */
167     private int onlineCount(Selector selector) {
168 
169         return 0;
170     }
171 
172     public static void main(String[] args) {
173         System.out.println("开始启动服务");
174         new Thread(new CharRoomServer()).start();
175         System.out.println("服务启动");
176     }
177 }

  1 import java.io.IOException;
  2 import java.net.InetSocketAddress;
  3 import java.nio.ByteBuffer;
  4 import java.nio.channels.SelectionKey;
  5 import java.nio.channels.Selector;
  6 import java.nio.channels.SocketChannel;
  7 import java.nio.charset.Charset;
  8 import java.util.Iterator;
  9 import java.util.Scanner;
 10 
 11 /**
 12  * @ClassName ChatRoomClient
 13  * @Description TODO
 14  * @Author hufeng8
 15  * @Date 2018/8/3 16:37
 16  * @Version 1.0
 17  */
 18 public class ChatRoomClient implements Runnable{
 19 
 20 
 21     private SocketChannel channel;
 22     private Selector selector;
 23     private boolean active = true;
 24     private Charset charset = Charset.forName("UTF-8");
 25 
 26     private String name = "";
 27 
 28     public ChatRoomClient() {
 29 
 30         try {
 31             selector = Selector.open();
 32             channel = SocketChannel.open(new InetSocketAddress("localhost",CharRoomServer.PORT_NUM));
 33             channel.configureBlocking(false);
 34             channel.register(selector, SelectionKey.OP_CONNECT|SelectionKey.OP_READ);
 35 
 36         } catch (IOException e) {
 37             e.printStackTrace();
 38         }
 39     }
 40 
 41     public void run() {
 42         System.out.println("客户端监听开始：");
 43         while (active) {
 44             try {
 45                 selector.select();
 46             } catch (IOException e) {
 47                 e.printStackTrace();
 48             }
 49 //            System.out.println(selector.selectedKeys().size());
 50             Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
 51             while (iterator.hasNext()) {
 52                 SelectionKey next = iterator.next();
 53                 iterator.remove();
 54                 if (channel == next.channel()) {
 55 //                    System.out.println("同一个对象");
 56                 }
 57 //                System.out.println("channel:"+channel+", sockchannel"+next.channel()+"连接事件："+next.isConnectable()+", "+next.isReadable()+", "+next.isWritable());
 58                 if (next.isConnectable()) {
 59                     try {
 60                         System.out.println("客户端连接事件");
 61                         SocketChannel sc = (SocketChannel) next.channel();
 62                         sc.configureBlocking(false);
 63                         sc.register(selector, SelectionKey.OP_READ);
 64                     } catch (IOException e) {
 65                         e.printStackTrace();
 66                     }
 67                 } else if (next.isWritable()) {
 68                     System.out.println("客户端写事件");
 69                 } else if (next.isReadable()) {
 70                     SocketChannel sc = (SocketChannel) next.channel();
 71                     if (channel == sc) {
 72                         System.out.println("同一个对象");
 73                     }
 74                     System.out.println("客户端读取事件");
 75                     ByteBuffer bf = ByteBuffer.allocate(2*1024);
 76                     int i = 0;
 77                     StringBuffer content = new StringBuffer();
 78                     try {
 79                         while ((i = sc.read(bf)) > 0) {
 80                             bf.flip();
 81                             content.append(charset.decode(bf));
 82                         }
 83                         System.out.println(content.toString());
 84 //                        next.interestOps(SelectionKey.OP_READ);
 85                     } catch (IOException e) {
 86                         e.printStackTrace();
 87                         next.cancel();
 88                         if (sc != null) {
 89                             try {
 90                                 sc.close();
 91                             } catch (IOException e1) {
 92                                 e1.printStackTrace();
 93                             }
 94                         }
 95                     }
 96                 }
 97             }
 98         }
 99     }
100 
101     public SocketChannel getChannel() {
102         return channel;
103     }
104 
105     public String getName() {
106         return name;
107     }
108 
109     public void setName(String name) {
110         this.name = name;
111     }
112 
113     public Charset getCharset() {
114         return charset;
115     }
116 
117     public void setCharset(Charset charset) {
118         this.charset = charset;
119     }
120 
121     public static void main(String[] args) {
122         ChatRoomClient chatRoomClient = new ChatRoomClient();
123         new Thread(chatRoomClient).start();
124         String name = chatRoomClient.getName();
125         SocketChannel sc = chatRoomClient.getChannel();
126         Charset charset = chatRoomClient.getCharset();
127         System.out.println("请输入：");
128         Scanner scan = new Scanner(System.in);
129         while (scan.hasNextLine()) {
130             String data = scan.next();
131             if ("".equals(data)) {
132                 continue;
133             } else if ("".equals(name)) {
134                 name = data;
135                 data = name + CharRoomServer.protocol;
136             } else {
137                 data = name + CharRoomServer.protocol + data;
138             }
139             try {
140                 sc.write(charset.encode(data));
141             } catch (IOException e) {
142 
143             }
144         }
145 
146     }
147 }