AsyncTimeServer实例

基于操作系统内核的服务器版本，与内核交互较多所以基本都是嵌套回调执行

1package aio;
2
3import java.io.IOException;
4import java.net.InetSocketAddress;
5import java.nio.ByteBuffer;
6import java.nio.channels.AsynchronousServerSocketChannel;
7import java.nio.channels.AsynchronousSocketChannel;
8import java.nio.channels.CompletionHandler;
9import java.util.concurrent.CountDownLatch;
10
11/**
12 * 启动服务端监听
13 * @author DXCyber409
14 */
15public class AsyncTimeServer {
16
17    public static void main(String[] args) {
18        AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(9999);
19        new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start();
20    }
21
22}
23
24/**
25 * 服务端
26 * @author DXCyber409
27 */
28class AsyncTimeServerHandler implements Runnable {
29
30    private int port;
31    
32    CountDownLatch latch;
33    AsynchronousServerSocketChannel asynchronousServerSocketChannel;
34
35    public AsyncTimeServerHandler(int port) {
36        this.port = port;
37        try {
38            asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
39            asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
40            System.out.println("The time server is start in port : " + port);
41        } catch (IOException ex) {
42            ex.printStackTrace();
43            System.exit(1);
44        }
45    }
46
47    @Override
48    public void run() {
49        latch = new CountDownLatch(1);  // CountDownLatch允许在计数为0之前挂起当前操作线程，防止主线程过早结束
50        doAccept();
51        try {
52            latch.await();
53        } catch (InterruptedException ex) {
54            ex.printStackTrace();
55        }
56    }
57    
58    public void doAccept() {
59        asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());
60    }
61}
62
63/**
64 * 连接事件完成处理回调器
65 * @author DXCyber409
66 */
67class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler>{
68
69    /**
70     * 连接成功事件回调
71     * @param result
72     * @param attachment 
73     */
74    @Override
75    public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
76        attachment.asynchronousServerSocketChannel.accept(attachment, this);
77        ByteBuffer buffer = ByteBuffer.allocate(1024);
78        result.read(buffer, buffer, new ReadCompletionHandler(result));
79    }
80
81    /**
82     * 连接失败事件回调
83     * @param exc
84     * @param attachment 
85     */
86    @Override
87    public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
88        exc.printStackTrace();
89        attachment.latch.countDown();
90    }
91    
92}
93
94/**
95 * 读取事件完成处理回调器
96 * @author DXCyber409
97 */
98class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer>{
99    
100    private AsynchronousSocketChannel channel;
101    
102    public ReadCompletionHandler(AsynchronousSocketChannel channel) {
103        if (this.channel == null) {
104            this.channel = channel;
105        }
106    }
107
108    /**
109     * 成功状态回调
110     * @param result
111     * @param attachment 
112     */
113    @Override
114    public void completed(Integer result, ByteBuffer attachment) {
115        attachment.flip();
116        byte[] body = new byte[attachment.remaining()];
117        attachment.get(body);
118        try {
119            String req = new String(body);
120            System.out.println("The time server receive order : " + req);
121            String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req) 
122                    ? new java.util.Date(System.currentTimeMillis()).toString() : "BAD ORDER";
123            doWrite(currentTime);
124        } catch (Exception e) {
125            e.printStackTrace();
126        }
127    }
128    
129    /**
130     * 分离出写模块
131     * @param response 
132     */
133    private void doWrite(String response) {
134        if (response != null && response.trim().length() > 0) {
135            byte[] bytes = response.getBytes();
136            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
137            writeBuffer.put(bytes);
138            writeBuffer.flip();
139            channel.write(writeBuffer, writeBuffer,
140                new CompletionHandler<Integer, ByteBuffer>() {
141                    
142                    /**
143                     * 成功状态回调
144                     * @param result
145                     * @param buffer 
146                     */
147                    @Override
148                    public void completed(Integer result, ByteBuffer buffer) {
149                        if (buffer.hasRemaining()) {  // 有可能写半包，若真如此则继续写
150                            channel.write(buffer, buffer, this);
151                        }
152                    }
153
154                    /**
155                     * 失败状态回调
156                     * @param exc
157                     * @param attachment 
158                     */
159                    @Override
160                    public void failed(Throwable exc, ByteBuffer attachment) {
161                        try {
162                            channel.close();
163                        } catch(IOException e) {
164                            e.printStackTrace();
165                        }
166                    }
167                }
168            );
169        }
170    }
171
172    /**
173     * 失败状态回调
174     * @param exc
175     * @param attachment 
176     */
177    @Override
178    public void failed(Throwable exc, ByteBuffer attachment) {
179        try {
180            this.channel.close();
181        } catch (Exception e) {
182            e.printStackTrace();
183        }
184    }
185    
186}