Java NIO简易聊天室

1. public abstract class Chat extends Observable { 
2.   
3.     public static final int SEV_ON = 0; 
4.     public static final int SEV_OFF = 1; 
5.     public static final int CLT_CONNECT = 2; 
6.     public static final int CLT_DISCONNECT = 3; 
7.     public static final int MSG_SEND = 4; 
8.     public static final int MSG_RECEIVE = 5; 
9.     public static final int ERROR = 6; 
10.   
11.     /** 缓存区大小 */ 
12.     protected static final int BUFFERSIZE = 1024 * 10; 
13.     /** 字符编码 */ 
14.     protected static final String CHARSET = "UTF-8"; 
15.   
16.     /** 字符编码器 */ 
17.     protected static CharsetEncoder encoder; 
18.     /** 字符解码器 */ 
19.     protected static CharsetDecoder decoder; 
20.   
21.     static { 
22.        encoder = Charset.forName(CHARSET).newEncoder(); 
23.        decoder = Charset.forName(CHARSET).newDecoder(); 
24.     } 
25.   
26.     /** 当前状态 */ 
27.     protected int status; 
28.   
29.     /** 获得当前状态 */ 
30.     public int getStatus() { 
31.        return status; 
32.     } 
33.   
34.     /** 
35.      * 通知状态改变 
36.      * @param status 状态 
37.      * @param arg 参数 
38.      */ 
39.     protected void notifyStateChanged(int status, Object arg) { 
40.        this.status = status; 
41.        notifyStateChanged(arg); 
42.     } 
43.   
44.     /** 
45.      * 通知状态改变 
46.      * @param arg 参数 
47.      */ 
48.     protected void notifyStateChanged(Object arg) { 
49.        setChanged(); 
50.        notifyObservers(arg); 
51.     } 
52.   
53. } 

1. public class ChatServer extends Chat implements Runnable { 
2.   
3.     private boolean isPrepared = false; 
4.     private ServerSocketChannel ssc; 
5.     private Selector selector; 
6.     private ArrayList<SelectionKey> serverKeyList; 
7.     private String receiveMessage; 
8.   
9.     /** 
10.      * 服务器构造函数 
11.      * 
12.      * @param port 端口 
13.      */ 
14.     public ChatServer(int port) { 
15.        try { 
16.            selector = Selector.open(); 
17.            ssc = ServerSocketChannel.open(); 
18.            ssc.configureBlocking(false); 
19.            ssc.socket().bind(new InetSocketAddress(port)); 
20.            ssc.register(selector, SelectionKey.OP_ACCEPT); 
21.            serverKeyList = new ArrayList<SelectionKey>(); 
22.            isPrepared = true; 
23.        } catch (IOException e) { 
24.            notifyStateChanged(ERROR, e); 
25.            e.printStackTrace(); 
26.        } 
27.     } 
28.   
29.     public void start() { 
30.        if (isPrepared) 
31.            new Thread(this).start(); 
32.     } 
33.   
34.     /** 针对同一个SelectionKey在一次写操作之前，后attach进去的消息会被覆盖。 */ 
35.     public void send(String msg, InetSocketAddress... toIps) { 
36.        notifyStateChanged(MSG_SEND, msg); 
37.        if (null == serverKeyList || serverKeyList.size() <= 0) { 
38.            return; 
39.        } 
40.        if (null != toIps && toIps.length >= 1) { 
41.            /* 发送给部分 */ 
42.            for (SelectionKey serverKey : serverKeyList) { 
43.               SocketChannel sc = (SocketChannel) serverKey.channel(); 
44.               SocketAddress ip = sc.socket().getRemoteSocketAddress(); 
45.               for (InetSocketAddress toIp : toIps) { 
46.                   if (toIp.equals(ip)) { 
47.                      serverKey.attach(msg); 
48.                      serverKey.interestOps(SelectionKey.OP_READ 
49.                             | SelectionKey.OP_WRITE); 
50.                      serverKey.selector().wakeup(); 
51.                      break; 
52.                   } 
53.               } 
54.            } 
55.        } else { 
56.            /* 发送给全部 */ 
57.            for (SelectionKey serverKey : serverKeyList) { 
58.               serverKey.attach(msg); 
59.               serverKey.interestOps(SelectionKey.OP_READ 
60.                      | SelectionKey.OP_WRITE); 
61.               serverKey.selector().wakeup(); 
62.            } 
63.        } 
64.     } 
65.   
66.     @Override 
67.     public void run() { 
68.        notifyStateChanged(SEV_ON, null); 
69.        try { 
70.            while (isPrepared) { 
71.               int keysCount = selector.select(); 
72.               if (keysCount < 1) { 
73.                   continue; 
74.               } 
75.               Set<SelectionKey> set = selector.selectedKeys(); 
76.               Iterator<SelectionKey> it = set.iterator(); 
77.               while (it.hasNext()) { 
78.                   SelectionKey key = it.next(); 
79.                   if (key.isAcceptable()) { 
80.                      doAccept(key); 
81.                   } 
82.                   if (key.isValid() && key.isReadable()) { 
83.                      doRead(key); 
84.                   } 
85.                   if (key.isValid() && key.isWritable()) { 
86.                      doWrite(key); 
87.                   } 
88.               } 
89.               set.clear(); 
90.            } 
91.        } catch (IOException e) { 
92.            e.printStackTrace(); 
93.        } finally { 
94.            // close(); 
95.            notifyStateChanged(SEV_OFF, null); 
96.        } 
97.     } 
98.   
99.     public void close() { 
100.        isPrepared = false; 
101.        try { 
102.            if (null != serverKeyList) { 
103.               for (SelectionKey key : serverKeyList) { 
104.                   key.channel().close(); 
105.               } 
106.            } 
107.            if (null != selector) { 
108.               selector.close(); 
109.            } 
110.            if (null != ssc) { 
111.               ssc.close(); 
112.            } 
113.        } catch (IOException e) { 
114.            e.printStackTrace(); 
115.        } 
116.     } 
117.   
118.     private void doAccept(SelectionKey key) { 
119.        ServerSocketChannel ssc = (ServerSocketChannel) key.channel(); 
120.        try { 
121.            SocketChannel sc = ssc.accept(); 
122.            sc.configureBlocking(false); 
123.            SelectionKey newKey = sc.register(selector, SelectionKey.OP_READ); 
124.            // newKey.attach(new ArrayList<String>()); 
125.  
126.            serverKeyList.add(newKey); 
127.            notifyStateChanged(CLT_CONNECT, sc.socket() 
128.                   .getRemoteSocketAddress()); 
129.        } catch (IOException e) { 
130.            e.printStackTrace(); 
131.        } 
132.     } 
133.   
134.     private void doRead(SelectionKey key) { 
135.        SocketChannel sc = (SocketChannel) key.channel(); 
136.        ByteBuffer bb = ByteBuffer.allocate(BUFFERSIZE); 
137.        StringBuffer sb = new StringBuffer(); 
138.        try { 
139.            int count = 0; 
140.            while ((count = sc.read(bb)) > 0) { 
141.               bb.flip(); 
142.               sb.append(decoder.decode(bb)); 
143.            } 
144.            if (count == -1) { 
145.               disconnect(key, sc); 
146.            } else { 
147.               receiveMessage = sb.toString().trim(); 
148.               notifyStateChanged(MSG_RECEIVE, sc.socket() 
149.                      .getRemoteSocketAddress()); 
150.   
151.            } 
152.        } catch (IOException e) { 
153.            disconnect(key, sc); 
154.            // e.printStackTrace(); 
155.  
156.        } 
157.     } 
158.   
159.     private void doWrite(SelectionKey key) { 
160.        SocketChannel sc = (SocketChannel) key.channel(); 
161.        String msg = (String) key.attachment(); 
162.        if (null == msg) { 
163.            key.interestOps(SelectionKey.OP_READ); 
164.            return; 
165.        } 
166.        try { 
167.            sc.write(encoder.encode(CharBuffer.wrap(msg))); 
168.        } catch (IOException e) { 
169.            disconnect(key, sc); 
170.            // e.printStackTrace(); 
171.  
172.        } 
173.        key.interestOps(SelectionKey.OP_READ); 
174.     } 
175.   
176.     /** 断开连接 */ 
177.     private void disconnect(SelectionKey key, SocketChannel sc) { 
178.        serverKeyList.remove(key); 
179.        notifyStateChanged(CLT_DISCONNECT, sc.socket().getRemoteSocketAddress()); 
180.  
181.        try { 
182.            key.cancel(); 
183.            sc.close(); 
184.        } catch (IOException e) { 
185.            e.printStackTrace(); 
186.        } 
187.     } 
188.   
189.     public String getReceiveMessage() { 
190.        return receiveMessage; 
191.     } 
192.   
193. } 

1. public class ChatClient extends Chat implements Runnable { 
2.   
3.     private boolean isPrepared = false; 
4.     private Selector selector; 
5.     private SelectionKey clientKey; 
6.     private InetSocketAddress address; 
7.   
8.     /** 
9.      * 客户端构造函数 
10.      * 
11.      * @param host 服务器地址 
12.      * @param port 服务器端口 
13.      */ 
14.     public ChatClient(String host, int port) { 
15.        address = new InetSocketAddress(host, port); 
16.        try { 
17.            selector = Selector.open(); 
18.        } catch (IOException e) { 
19.            notifyStateChanged(ERROR, e); 
20.            e.printStackTrace(); 
21.        } 
22.     } 
23.   
24.     public void start() { 
25.        new Thread(this).start(); 
26.     } 
27.   
28.     public void send(String msg) { 
29.        notifyStateChanged(MSG_SEND, msg); 
30.        if (null == clientKey) { 
31.            return; 
32.        } 
33.        clientKey.attach(msg); 
34.        clientKey.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE); 
35.        clientKey.selector().wakeup(); 
36.     } 
37.   
38.     @Override 
39.     public void run() { 
40.        try { 
41.            SocketChannel sc = SocketChannel.open(); 
42.            sc.configureBlocking(false); 
43.            sc.connect(address); 
44.            clientKey = sc.register(selector, SelectionKey.OP_CONNECT); 
45.            isPrepared = true; 
46.            while (isPrepared) { 
47.               int keysCount = selector.select(); 
48.               if (keysCount < 1) { 
49.                   continue; 
50.               } 
51.               Set<SelectionKey> set = selector.selectedKeys(); 
52.               Iterator<SelectionKey> it = set.iterator(); 
53.               while (it.hasNext()) { 
54.                   SelectionKey key = it.next(); 
55.                   if (key.isConnectable()) { 
56.                      doConnect(key); 
57.                   } 
58.                   if (key.isValid() && key.isReadable()) { 
59.                      doRead(key); 
60.                   } 
61.                   if (key.isValid() && key.isWritable()) { 
62.                      doWrite(key); 
63.                   } 
64.               } 
65.               set.clear(); 
66.            } 
67.        } catch (IOException e) { 
68.            notifyStateChanged(ERROR, e); 
69.            e.printStackTrace(); 
70.        } finally { 
71.            // close(); 
72.            notifyStateChanged(CLT_DISCONNECT, null); 
73.        } 
74.     } 
75.   
76.     public void close() { 
77.        isPrepared = false; 
78.        try { 
79.            if (null != clientKey) { 
80.               clientKey.channel().close(); 
81.            } 
82.            if (null != selector) { 
83.               selector.close(); 
84.            } 
85.        } catch (IOException e) { 
86.            e.printStackTrace(); 
87.        } 
88.     } 
89.   
90.     private void doConnect(SelectionKey key) { 
91.        SocketChannel sc = (SocketChannel) key.channel(); 
92.        try { 
93.            // http://www.velocityreviews.com/forums/t145075-whats-the-proper-way-to-use-socketchannel-finishconnect.html 
94.  
95.            sc.finishConnect(); 
96.            key.interestOps(SelectionKey.OP_READ); 
97.            notifyStateChanged(CLT_CONNECT, null); 
98.        } catch (IOException e) { 
99.            disconnect(key); 
100.            e.printStackTrace(); 
101.        } 
102.     } 
103.   
104.     private void doRead(SelectionKey key) { 
105.        SocketChannel sc = (SocketChannel) key.channel(); 
106.        ByteBuffer bb = ByteBuffer.allocate(BUFFERSIZE); 
107.        StringBuffer sb = new StringBuffer(); 
108.        try { 
109.            int count = 0; 
110.            while ((count = sc.read(bb)) > 0) { 
111.               bb.flip(); 
112.               sb.append(decoder.decode(bb)); 
113.            } 
114.            if (count == -1) { 
115.               disconnect(key); 
116.            } else { 
117.               notifyStateChanged(MSG_RECEIVE, sb.toString().trim()); 
118.            } 
119.        } catch (IOException e) { 
120.            disconnect(key); 
121.            e.printStackTrace(); 
122.        } 
123.     } 
124.   
125.     private void doWrite(SelectionKey key) { 
126.        SocketChannel sc = (SocketChannel) key.channel(); 
127.        String msg = (String) key.attachment(); 
128.        if (null == msg) { 
129.            key.interestOps(SelectionKey.OP_READ); 
130.            return; 
131.        } 
132.        try { 
133.            sc.write(encoder.encode(CharBuffer.wrap(msg))); 
134.        } catch (IOException e) { 
135.            disconnect(key); 
136.            e.printStackTrace(); 
137.        } 
138.        key.interestOps(SelectionKey.OP_READ); 
139.     } 
140.   
141.     /** 断开连接 */ 
142.     private void disconnect(SelectionKey key) { 
143.        notifyStateChanged(CLT_DISCONNECT, null); 
144.        try { 
145.            key.cancel(); 
146.            key.channel().close(); 
147.        } catch (IOException e) { 
148.            e.printStackTrace(); 
149.        } 
150.     } 
151.   
152. }