代码实现

1 // Reactor線程
2 package server;
3
4 import java.io.IOException;
5 import java.net.InetSocketAddress;
6 import java.nio.channels.SelectionKey;
7 import java.nio.channels.Selector;
8 import java.nio.channels.ServerSocketChannel;
9 import java.util.Iterator;
10 import java.util.Set;
11
12 public class TCPReactor implements Runnable {
13
14 private final ServerSocketChannel ssc;
15 private final Selector selector;
16
17 public TCPReactor(int port) throws IOException {
18 selector = Selector.open();
19 ssc = ServerSocketChannel.open();
20 InetSocketAddress addr = new InetSocketAddress(port);
21 ssc.socket().bind(addr); // 在ServerSocketChannel綁定監聽端口
22 ssc.configureBlocking(false); // 設置ServerSocketChannel為非阻塞
23 SelectionKey sk = ssc.register(selector, SelectionKey.OP_ACCEPT); // ServerSocketChannel向selector註冊一個OP_ACCEPT事件，然後返回該通道的key
24 sk.attach(new Acceptor(selector, ssc)); // 給定key一個附加的Acceptor對象
25 }
26
27 @Override
28 public void run() {
29 while (!Thread.interrupted()) { // 在線程被中斷前持續運行
30 System.out.println("Waiting for new event on port: " + ssc.socket().getLocalPort() + "…");
31 try {
32 if (selector.select() == 0) // 若沒有事件就緒則不往下執行
33 continue;
34 } catch (IOException e) {
35 // TODO Auto-generated catch block
36 e.printStackTrace();
37 }
38 Set selectedKeys = selector.selectedKeys(); // 取得所有已就緒事件的key集合
39 Iterator it = selectedKeys.iterator();
40 while (it.hasNext()) {
41 dispatch((SelectionKey) (it.next())); // 根據事件的key進行調度
42 it.remove();
43 }
44 }
45 }
46
47 /*
48 * name: dispatch(SelectionKey key)
49 * description: 調度方法，根據事件綁定的對象開新線程
50 */
51 private void dispatch(SelectionKey key) {
52 Runnable r = (Runnable) (key.attachment()); // 根據事件之key綁定的對象開新線程
53 if (r != null)
54 r.run();
55 }
56
57 }

1 // 接受連線請求線程
2 package server;
3
4 import java.io.IOException;
5 import java.nio.channels.SelectionKey;
6 import java.nio.channels.Selector;
7 import java.nio.channels.ServerSocketChannel;
8 import java.nio.channels.SocketChannel;
9
10 public class Acceptor implements Runnable {
11
12 private final ServerSocketChannel ssc;
13 private final Selector selector;
14
15 public Acceptor(Selector selector, ServerSocketChannel ssc) {
16 this.ssc=ssc;
17 this.selector=selector;
18 }
19
20 @Override
21 public void run() {
22 try {
23 SocketChannel sc= ssc.accept(); // 接受client連線請求
24 System.out.println(sc.socket().getRemoteSocketAddress().toString() + " is connected.");
25
26 if(sc!=null) {
27 sc.configureBlocking(false); // 設置為非阻塞
28 SelectionKey sk = sc.register(selector, SelectionKey.OP_READ); // SocketChannel向selector註冊一個OP_READ事件，然後返回該通道的key
29 selector.wakeup(); // 使一個阻塞住的selector操作立即返回
30 sk.attach(new TCPHandler(sk, sc)); // 給定key一個附加的TCPHandler對象
31 }
32
33 } catch (IOException e) {
34 // TODO Auto-generated catch block
35 e.printStackTrace();
36 }
37 }
38
39
40 }

1 // Handler線程
2 package server;
3
4 import java.io.IOException;
5 import java.nio.ByteBuffer;
6 import java.nio.channels.SelectionKey;
7 import java.nio.channels.SocketChannel;
8 import java.util.concurrent.LinkedBlockingQueue;
9 import java.util.concurrent.ThreadPoolExecutor;
10 import java.util.concurrent.TimeUnit;
11
12 public class TCPHandler implements Runnable {
13
14 private final SelectionKey sk;
15 private final SocketChannel sc;
16
17 int state;
18
19 public TCPHandler(SelectionKey sk, SocketChannel sc) {
20 this.sk = sk;
21 this.sc = sc;
22 state = 0; // 初始狀態設定為READING
23 }
24
25 @Override
26 public void run() {
27 try {
28 if (state == 0)
29 read(); // 讀取網絡數據
30 else
31 send(); // 發送網絡數據
32
33 } catch (IOException e) {
34 System.out.println("[Warning!] A client has been closed.");
35 closeChannel();
36 }
37 }
38
39 private void closeChannel() {
40 try {
41 sk.cancel();
42 sc.close();
43 } catch (IOException e1) {
44 e1.printStackTrace();
45 }
46 }
47
48 private synchronized void read() throws IOException {
49 // non-blocking下不可用Readers，因為Readers不支援non-blocking
50 byte[] arr = new byte[1024];
51 ByteBuffer buf = ByteBuffer.wrap(arr);
52
53 int numBytes = sc.read(buf); // 讀取字符串
54 if(numBytes == -1)
55 {
56 System.out.println("[Warning!] A client has been closed.");
57 closeChannel();
58 return;
59 }
60 String str = new String(arr); // 將讀取到的byte內容轉為字符串型態
61 if ((str != null) && !str.equals(" ")) {
62 process(str); // 邏輯處理
63 System.out.println(sc.socket().getRemoteSocketAddress().toString()
64 + " > " + str);
65 state = 1; // 改變狀態
66 sk.interestOps(SelectionKey.OP_WRITE); // 通過key改變通道註冊的事件
67 sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回
68 }
69 }
70
71 private void send() throws IOException {
72 // get message from message queue
73
74 String str = "Your message has sent to "
75 + sc.socket().getLocalSocketAddress().toString() + "\r\n";
76 ByteBuffer buf = ByteBuffer.wrap(str.getBytes()); // wrap自動把buf的position設為0，所以不需要再flip()
77
78 while (buf.hasRemaining()) {
79 sc.write(buf); // 回傳給client回應字符串，發送buf的position位置到limit位置為止之間的內容
80 }
81
82 state = 0; // 改變狀態
83 sk.interestOps(SelectionKey.OP_READ); // 通過key改變通道註冊的事件
84 sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回
85 }
86
87 void process(String str) {
88 // do process(decode, logically process, encode)..
89 // ..
90 }
91 }

1 package server;
2
3 import java.io.IOException;
4
5 public class Main {
6
7
8 public static void main(String[] args) {
9 // TODO Auto-generated method stub
10 try {
11 TCPReactor reactor = new TCPReactor(1333);
12 reactor.run();
13 } catch (IOException e) {
14 // TODO Auto-generated catch block
15 e.printStackTrace();
16 }
17 }
18
19 }

客户端代码

1 package main.pkg;
2
3 import java.io.BufferedReader;
4 import java.io.IOException;
5 import java.io.InputStreamReader;
6 import java.io.PrintWriter;
7 import java.net.Socket;
8 import java.net.UnknownHostException;
9
10 public class Client {
11
12 /**
13 * @param args
14 */
15 public static void main(String[] args) {
16 // TODO Auto-generated method stub
17 String hostname=args[0];
18 int port = Integer.parseInt(args[1]);
19 //String hostname="127.0.0.1";
20 //int port=1333;
21
22 System.out.println("Connecting to "+ hostname +":"+port);
23 try {
24 Socket client = new Socket(hostname, port); // 連接至目的地
25 System.out.println("Connected to "+ hostname);
26
27 PrintWriter out = new PrintWriter(client.getOutputStream());
28 BufferedReader in = new BufferedReader(new InputStreamReader(client.getInputStream()));
29 BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in));
30 String input;
31
32 while((input=stdIn.readLine()) != null) { // 讀取輸入
33 out.println(input); // 發送輸入的字符串
34 out.flush(); // 強制將緩衝區內的數據輸出
35 if(input.equals("exit"))
36 {
37 break;
38 }
39 System.out.println("server: "+in.readLine());
40 }
41 client.close();
42 System.out.println("client stop.");
43 } catch (UnknownHostException e) {
44 // TODO Auto-generated catch block
45 System.err.println("Don't know about host: " + hostname);
46 } catch (IOException e) {
47 // TODO Auto-generated catch block
48 System.err.println("Couldn't get I/O for the socket connection");
49 }
50
51 }
52
53 }