本章要點(diǎn)：
- 什么是netty
- 基本概念
- IO模型
1.1 什么是netty
Netty是由JBOSS提供的一個(gè)java開源框架。Netty提供異步的、事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用程序框架和工具，用以快速開發(fā)高性能、高可靠性的網(wǎng)絡(luò)服務(wù)器和客戶端程序。
也就是說，Netty 是一個(gè)基于NIO的客戶、服務(wù)器端編程框架，使用Netty 可以確保你快速和簡(jiǎn)單的開發(fā)出一個(gè)網(wǎng)絡(luò)應(yīng)用，例如實(shí)現(xiàn)了某種協(xié)議的客戶，服務(wù)端應(yīng)用。Netty相當(dāng)簡(jiǎn)化和流線化了網(wǎng)絡(luò)應(yīng)用的編程開發(fā)過程，例如，TCP和UDP的socket服務(wù)開發(fā)。----來自百度百科
要想真正的了解netty，首先要了解一些基本概念和IO模型。
1.2 基本概念
IO 模型可分為：同步阻塞IO（Blocking IO）、同步非阻塞IO（Non-blocking IO）、IO多路復(fù)用（IO Multiplexing）、 異步IO（Asynchronous IO）。要想明白這幾種類型的區(qū)別，首先要搞清楚以下幾個(gè)概念：
1.2.1 異步與同步
同步與異步是指消息通信機(jī)制：
- 同步：當(dāng)你調(diào)用其他的服務(wù)或者某個(gè)方法時(shí)，在這個(gè)服務(wù)或者方法返回結(jié)果之前，你不能夠做別的事情，只能一直等待。
- 異步： 當(dāng)調(diào)用其他服務(wù)或者方法時(shí),調(diào)用者立即返回結(jié)果。但是實(shí)際真正的結(jié)果只有處理這個(gè)調(diào)用的部件在完成后,通過狀態(tài)、通知和回調(diào)來通知調(diào)用者。
同步與異步是對(duì)應(yīng)的，它們是線程之間的關(guān)系，兩個(gè)線程之間要么是同步的，要么是異步的。
1.2.2 阻塞與非阻塞
阻塞和非阻塞關(guān)注的是程序在等待調(diào)用結(jié)果（消息，返回值）時(shí)的狀態(tài)，是數(shù)據(jù)處理的方式。
- 阻塞：當(dāng)調(diào)用時(shí)會(huì)將此線程會(huì)掛起，直到線程持續(xù)等待資源中數(shù)據(jù)準(zhǔn)備完成，返回響應(yīng)結(jié)果。
- 非阻塞：非阻塞和阻塞的概念相對(duì)應(yīng),指在不能立刻得到結(jié)果之前,該函數(shù)不會(huì)阻塞當(dāng)前線程,而會(huì)立刻返回。
阻塞與非阻塞是對(duì)同一個(gè)線程來說的，在某個(gè)時(shí)刻，線程要么處于阻塞，要么處于非阻塞。
1.3 IO模型
1.3.1 傳統(tǒng)的BIO
傳統(tǒng)的BIO是一種同步阻塞的IO，在IO讀寫該線程都會(huì)進(jìn)行阻塞，無法進(jìn)行其他操作。該模式是一問一答的模式，所以服務(wù)端的線程個(gè)數(shù)和客戶端的并發(fā)訪問數(shù)是1:1的關(guān)系，由于線程是JVM非常寶貴的系統(tǒng)資源，當(dāng)線程數(shù)越來越多時(shí)，系統(tǒng)會(huì)急劇下降，甚至宕機(jī)。
BIO代碼示例如下：
服務(wù)端：
package com.bj58.wuxian.bio;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
public class TimeServer {
public static void main(String[] args) throws IOException {
ServerSocket serverSocket=null;
try {
serverSocket=new ServerSocket(8888);
Socket socket=null;
while(true){
socket=serverSocket.accept();
new Thread(new TimeServerHandler(socket)).start();
}
}finally {
if(serverSocket!=null){
serverSocket.close();
}
}
}
}
服務(wù)端處理器：
package com.bj58.wuxian.bio;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.Socket;
public class TimeServerHandler implements Runnable{
Socket socket;
public TimeServerHandler(Socket socket) {
super();
this.socket = socket;
}
@Override
public void run() {
BufferedReader bufferedReader=null;
String info=null;
String currentTime = null;
BufferedWriter bufferedWriter=null;
try {
bufferedReader=new BufferedReader(new InputStreamReader(socket.getInputStream(), "utf-8"));
bufferedWriter=new BufferedWriter(new OutputStreamWriter(socket.getOutputStream(),"utf-8"));
while (true) {
info=bufferedReader.readLine();
if(info==null){
break;
}
System.out.println("request:"+info);
currentTime = "QUERY TIME ORDER".equalsIgnoreCase(info) ? new java.util.Date(
System.currentTimeMillis()).toString() : "BAD ORDER";
bufferedWriter.write(currentTime+"\r\n");
bufferedWriter.flush();
}
} catch (IOException e) {
if (bufferedReader != null) {
try {
bufferedReader.close();
} catch (IOException e1) {
e1.printStackTrace();
}
}
if (bufferedWriter != null) {
try {
bufferedWriter.close();
} catch (IOException e1) {
e1.printStackTrace();
}
bufferedWriter = null;
}
if (this.socket != null) {
try {
this.socket.close();
} catch (IOException e1) {
e1.printStackTrace();
}
this.socket = null;
}
}
}
}
客戶端：
package com.bj58.wuxian.bio;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.Socket;
public class TimeClient {
public static void main(String[] args) {
Socket socket = null;
BufferedReader bufferedReader=null;
BufferedWriter bufferedWriter=null;
try {
socket = new Socket("127.0.0.1", 8888);
bufferedReader=new BufferedReader(new InputStreamReader(socket.getInputStream(), "utf-8"));
bufferedWriter=new BufferedWriter(new OutputStreamWriter(socket.getOutputStream(),"utf-8"));
bufferedWriter.write("QUERY TIME ORDER"+"\r\n");
bufferedWriter.flush();
String resp=bufferedReader.readLine();
System.out.println("Now is:"+resp);
} catch (Exception e) {
e.printStackTrace();
}finally {
if (bufferedWriter != null) {
try {
bufferedWriter.close();
} catch (IOException e) {
e.printStackTrace();
}
bufferedWriter = null;
}
if (bufferedReader != null) {
try {
bufferedReader.close();
} catch (IOException e) {
e.printStackTrace();
}
bufferedReader = null;
}
if (socket != null) {
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
socket = null;
}
}
}
}
1.3.2 偽異步IO模型
為了解決同步阻塞的一個(gè)線程一個(gè)連接的弊端，可以對(duì)他進(jìn)行優(yōu)化，即采用線程池和任務(wù)隊(duì)列可以實(shí)現(xiàn)一種偽異步的IO。
示例代碼如下：
server端運(yùn)用線程池：
package com.bj58.wuxian.pio;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import com.bj58.wuxian.bio.TimeServerHandler;
public class TimeServer {
public static void main(String[] args) throws IOException {
ServerSocket server = null;
try {
server = new ServerSocket(8888);
Socket socket = null;
TimeServerHandlerExecutePool singleExecutor = new TimeServerHandlerExecutePool(
50, 10000);
while (true) {
socket = server.accept();
singleExecutor.execute(new TimeServerHandler(socket));
}
} finally {
if (server != null) {
System.out.println("The time server close");
server.close();
server = null;
}
}
}
}
線程池：
package com.bj58.wuxian.pio;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class TimeServerHandlerExecutePool {
private ExecutorService executor;
public TimeServerHandlerExecutePool(int maxPoolSize, int queueSize) {
executor = new ThreadPoolExecutor(Runtime.getRuntime()
.availableProcessors(), maxPoolSize, 120L, TimeUnit.SECONDS,
new ArrayBlockingQueue<java.lang.Runnable>(queueSize));
}
public void execute(java.lang.Runnable task) {
executor.execute(task);
}
}
1.3.3 NIO模型
NIO是在JDK1.4引入的,它彌補(bǔ)了BIO的不足，NIO和BIO之間的最大的區(qū)別是，IO是面向流的，NIO是面向緩沖區(qū)的。它在標(biāo)準(zhǔn)的java代碼中提供了高速度的、面向塊（即緩沖區(qū)）的IO。NIO主要有三大核心部分：Channel(通道)，Buffer(緩沖區(qū)), Selector（多路復(fù)用器）。 下面我們了解一下NIO的類庫中的一些相關(guān)概念。
1.3.3.1 Channel(通道)：
Channel是一個(gè)通道并且是雙向的，可以通過它讀取和寫入數(shù)據(jù)。主要有以下：
FileChannel:操作文件的Channel
SocketChannel:操作Socket客戶端的Channel
ServerSocketChannel:操作Socket服務(wù)端的Channel
DatagramChannel: 操作UDP的Channel
1.3.3.2 Buffer(緩沖區(qū))：
Buffer顧名思義，它是一個(gè)緩沖區(qū)，實(shí)際上是一個(gè)容器，一個(gè)連續(xù)數(shù)組，它包括了一些要寫入或者要讀出的數(shù)據(jù)。在NIO中所有的數(shù)據(jù)都是用緩沖區(qū)來處理的。任何時(shí)候訪問NIO中的數(shù)據(jù)，都是通過緩沖區(qū)進(jìn)行操作的。
- capacity：作為一個(gè)內(nèi)存塊，Buffer有固定的容量，即“capacity”。一旦Buffer滿了，需要將其清空（通過讀數(shù)據(jù)或者清楚數(shù)據(jù)）才能繼續(xù)寫數(shù)據(jù)。
- position：當(dāng)你寫數(shù)據(jù)到Buffer中時(shí)，position表示當(dāng)前的位置。初始值為0，當(dāng)寫入數(shù)據(jù)時(shí)，position會(huì)向前移動(dòng)到下一個(gè)可插入數(shù)據(jù)的Buffer單元。position最大可為capacity-1。當(dāng)讀取數(shù)據(jù)時(shí)，也是從某個(gè)特定位置讀，將Buffer從寫模式切換到讀模式，position會(huì)被重置為0。當(dāng)從Buffer的position處讀取一個(gè)字節(jié)數(shù)據(jù)后，position向前移動(dòng)到下一個(gè)可讀的位置。
- limit：在寫模式下，Buffer的limit表示你最多能往Buffer里寫多少數(shù)據(jù)。 寫模式下，limit等于Buffer的capacity。當(dāng)切換Buffer到讀模式時(shí)， limit表示你最多能讀到多少數(shù)據(jù)。因此，當(dāng)切換Buffer到讀模式時(shí)，limit會(huì)被設(shè)置成寫模式下的position值。換句話說，你能讀到之前寫入的所有數(shù)據(jù)（limit被設(shè)置成已寫數(shù)據(jù)的數(shù)量，這個(gè)值在寫模式下就是position）
Buffer的分配：對(duì)Buffer對(duì)象的操作必須首先進(jìn)行分配，Buffer提供一個(gè)allocate(int capacity)方法分配一個(gè)指定字節(jié)大小的對(duì)象。
向Buffer中寫數(shù)據(jù)：寫數(shù)據(jù)到Buffer中有兩種方式：
- 從channel寫到Buffer
int bytes = channel.read(buf); //將channel中的數(shù)據(jù)讀取到buf中
- 通過Buffer的put()方法寫到Buffer
buf.put(byte); //將數(shù)據(jù)通過put()方法寫入到buf中
- flip()方法：將Buffer從寫模式切換到讀模式，調(diào)用flip()方法會(huì)將position設(shè)置為0，并將limit設(shè)置為之前的position的值。
從Buffer中讀數(shù)據(jù)：從Buffer中讀數(shù)據(jù)有兩種方式：
- 從Buffer讀取數(shù)據(jù)到Channel
int bytes = channel.write(buf); //將buf中的數(shù)據(jù)讀取到channel中
- 通過Buffer的get()方法讀取數(shù)據(jù)
byte bt = buf.get(); //從buf中讀取一個(gè)byte
rewind()方法：Buffer.rewind()方法將position設(shè)置為0，使得可以重讀Buffer中的所有數(shù)據(jù)，limit保持不變。
Buffer中的數(shù)據(jù)，讀取完成后，依然保存在Buffer中，可以重復(fù)讀取。clear()與compact()方法：一旦讀完Buffer中的數(shù)據(jù)，需要讓Buffer準(zhǔn)備好再次被寫入，可以通過clear()或compact()方法完成。如果調(diào)用的是clear()方法，position將被設(shè)置為0，limit設(shè)置為capacity的值。但是Buffer并未被清空，只是通過這些標(biāo)記告訴我們可以從哪里開始往Buffer中寫入多少數(shù)據(jù)。如果Buffer中還有一些未讀的數(shù)據(jù)，調(diào)用clear()方法將被"遺忘 "。compact()方法將所有未讀的數(shù)據(jù)拷貝到Buffer起始處，然后將position設(shè)置到最后一個(gè)未讀元素的后面，limit屬性依然設(shè)置為capacity。可以使得Buffer中的未讀數(shù)據(jù)還可以在后續(xù)中被使用。
mark()與reset()方法：通過調(diào)用Buffer.mark()方法可以標(biāo)記一個(gè)特定的position，之后可以通過調(diào)用Buffer.reset()恢復(fù)到這個(gè)position上。
每一個(gè)Java基本類型（除了Boolean類型外）都對(duì)應(yīng)一種緩沖區(qū)：ByteBuffer，CharBuffer，ShortBuffer，IntBuffer，LongBuffer，F(xiàn)loatBuffer，DoubleBuffer。
1.3.3.3 Selector（多路復(fù)用器）
多路選擇器提供選擇已經(jīng)就緒的任務(wù)的能力。Selector不斷的輪詢注冊(cè)在其上的Channel，如果某個(gè)Channel上面有新的TCP連接接入、讀和寫事件，這個(gè)Channel就處于就緒狀態(tài)，會(huì)被Selector輪詢出來，然后通過SelectionKey可以獲取就緒的Channel集合，Selector可以監(jiān)聽狀態(tài)有：Connect、Accept、Read、Write，當(dāng)監(jiān)聽到某一Channel的某個(gè)狀態(tài)時(shí)，才允許對(duì)Channel進(jìn)行相應(yīng)的操作。
- Connect：某一個(gè)客戶端連接成功后
- Accept：準(zhǔn)備好進(jìn)行連接
- Read:可讀
- Write:可寫
NIO時(shí)間服務(wù)實(shí)例：
package com.bj58.wuxian.nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
public class MultiplexerTimeServer implements Runnable {
private Selector selector;
private ServerSocketChannel serverSocketChannel;
private volatile boolean stop;
public MultiplexerTimeServer(int port) {
try {
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.socket().bind(new InetSocketAddress(port), 1024);
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("********start*******");
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
@Override
public void run() {
while (!stop) {
try {
selector.select(1000);
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
SelectionKey key = null;
while (iterator.hasNext()) {
key = iterator.next();
iterator.remove();
try {
handleInput(key);
} catch (Exception e) {
if (key != null) {
key.cancel();
if (key.channel() != null)
key.channel().close();
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
private void handleInput(SelectionKey key) throws IOException {
if (key.isValid()) {
if (key.isAcceptable()) {
// 得到一個(gè)連接
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
SocketChannel sc = ssc.accept();
sc.configureBlocking(false);
// 把連接注冊(cè)到選擇器上
sc.register(selector, SelectionKey.OP_READ);
}
if (key.isReadable()) {
SocketChannel sc = (SocketChannel) key.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int readBytes = sc.read(readBuffer);
if (readBytes > 0) {
readBuffer.flip();
byte[] bytes = new byte[readBuffer.remaining()];
readBuffer.get(bytes);
String body = new String(bytes, "utf-8");
System.out.println("The time server receive order : " + body);
String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)
? new java.util.Date(System.currentTimeMillis()).toString() : "BAD ORDER";
doWrite(sc, currentTime);
} else if (readBytes < 0) {
// 對(duì)端鏈路關(guān)閉
key.cancel();
sc.close();
} else {
;// 讀到0字節(jié)，忽略
}
}
}
}
private void doWrite(SocketChannel sc, String response) throws IOException {
if (response != null && response.trim().length() > 0) {
byte[] bytes = response.getBytes();
ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
buffer.put(bytes);
buffer.flip();
sc.write(buffer);
}
}
public void stop() {
this.stop = true;
}
}
package com.bj58.wuxian.nio;
public class TimeServer {
public static void main(String[] args) {
MultiplexerTimeServer timeServer= new MultiplexerTimeServer(8888);
new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();
}
}
客戶端處理器：
package com.bj58.wuxian.nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
public class TimeClientHandle implements Runnable {
private String host;
private int port;
private Selector selector;
private SocketChannel socketChannel;
private volatile boolean stop;
public TimeClientHandle(String host, int port) {
this.host = host;
this.port = port;
try {
selector=Selector.open();
socketChannel=SocketChannel.open();
socketChannel.configureBlocking(false);
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
@Override
public void run() {
try {
doConnect();
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
while(!stop){
try {
selector.select(1000);
Set<SelectionKey> selectionKeys =selector.selectedKeys();
Iterator<SelectionKey> iterator=selectionKeys.iterator();
SelectionKey key=null;
while(iterator.hasNext()){
key=iterator.next();
iterator.remove();
try{
handleInput(key);
}catch (Exception e) {
if (key != null) {
key.cancel();
if (key.channel() != null)
key.channel().close();
}
}
}
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
}
private void handleInput(SelectionKey key) throws IOException {
if (key.isValid()) {
// 判斷是否連接成功
SocketChannel sc = (SocketChannel) key.channel();
if (key.isConnectable()) {
if (sc.finishConnect()) {
sc.register(selector, SelectionKey.OP_READ);
doWrite(sc);
} else
System.exit(1);// 連接失敗，進(jìn)程退出
}
if (key.isReadable()) {
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int readBytes = sc.read(readBuffer);
if (readBytes > 0) {
readBuffer.flip();
byte[] bytes = new byte[readBuffer.remaining()];
readBuffer.get(bytes);
String body = new String(bytes, "UTF-8");
System.out.println("Now is : " + body);
this.stop = true;
} else if (readBytes < 0) {
// 對(duì)端鏈路關(guān)閉
key.cancel();
sc.close();
} else
; // 讀到0字節(jié)，忽略
}
}
}
private void doConnect() throws IOException {
// 如果直接連接成功，則注冊(cè)到多路復(fù)用器上，發(fā)送請(qǐng)求消息，讀應(yīng)答
if(socketChannel.connect(new InetSocketAddress(host, port))){
socketChannel.register(selector, SelectionKey.OP_READ);
doWrite(socketChannel);
}else{
socketChannel.register(selector, SelectionKey.OP_CONNECT);
}
}
private void doWrite(SocketChannel socketChannel) throws IOException {
byte[] req = "QUERY TIME ORDER".getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
writeBuffer.put(req);
writeBuffer.flip();
socketChannel.write(writeBuffer);
if (!writeBuffer.hasRemaining())
System.out.println("Send order 2 server succeed.");
}
}
package com.bj58.wuxian.nio;
public class TimeClient {
public static void main(String[] args) {
new Thread(new TimeClientHandle("127.0.0.1", 8888), "TimeClient-001")
.start();
}
}
是不是感覺很繁瑣，哈哈哈~~~~~~
1.3.4 AIO 模型
NIO2.0引入了新的異步通道的概念，并提供了異步文件通道和異步套接字通道的實(shí)現(xiàn)。異步通道提供兩種方式獲取操作結(jié)果。
- 通過java.util.concurrent.Future類來表示異步操作的結(jié)果
- 在執(zhí)行異步操作的時(shí)候傳入一個(gè)java.nio.channels.Channel
java.nio.channels.CompletionHandler接口的實(shí)現(xiàn)類作為操作完成的回調(diào)。
NIO2.0的異步套接字通道是真正的異步非阻塞IO，它不需要通過多路復(fù)用器(Selector)對(duì)注冊(cè)的通道進(jìn)行輪詢操作即可實(shí)現(xiàn)異步讀寫，從而簡(jiǎn)化了NIO的變成模型。
其實(shí)也不簡(jiǎn)單，哈哈哈~~~~~
實(shí)例如下：
Server端：
package com.bj58.wuxian.aio;
import java.io.IOException;
public class TimeServer {
public static void main(String[] args) throws IOException {
AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(8888);
new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start();
}
}
異步時(shí)間服務(wù)器的處理類：
package com.bj58.wuxian.aio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.util.concurrent.CountDownLatch;
public class AsyncTimeServerHandler implements Runnable {
private int port;
CountDownLatch latch;
AsynchronousServerSocketChannel asynchronousServerSocketChannel;
public AsyncTimeServerHandler(int port) {
this.port = port;
try {
asynchronousServerSocketChannel=AsynchronousServerSocketChannel.open();
asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
System.out.println("The time server is start in port : " + port);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
latch=new CountDownLatch(1);
doAccept();
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private void doAccept() {
asynchronousServerSocketChannel.accept(this,new AcceptCompletionHandler());
}
}
package com.bj58.wuxian.aio;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler> {
@Override
public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
attachment.asynchronousServerSocketChannel.accept(attachment,this);
ByteBuffer buffer=ByteBuffer.allocate(1024);
result.read(buffer, buffer, new ReadCompletionHandler(result));
}
@Override
public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
exc.printStackTrace();
attachment.latch.countDown();
}
}
客戶端：
package com.bj58.wuxian.aio;
import java.io.IOException;
public class TimeServer {
public static void main(String[] args) throws IOException {
AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(8888);
new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start();
}
}
package com.bj58.wuxian.aio;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;
public class AsyncTimeClientHandler implements CompletionHandler<Void, AsyncTimeClientHandler>, Runnable {
private AsynchronousSocketChannel client;
private String host;
private int port;
private CountDownLatch latch;
public AsyncTimeClientHandler(String host, int port) {
this.host = host;
this.port = port;
try {
client = AsynchronousSocketChannel.open();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
latch = new CountDownLatch(1);
client.connect(new InetSocketAddress(host, port), this, this);
try {
latch.await();
} catch (InterruptedException e1) {
e1.printStackTrace();
}
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void completed(Void result, AsyncTimeClientHandler attachment) {
byte[] req = "QUERY TIME ORDER".getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
writeBuffer.put(req);
writeBuffer.flip();
client.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer buffer) {
if (buffer.hasRemaining()) {
client.write(buffer, buffer, this);
} else {
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
client.read(readBuffer, readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer buffer) {
buffer.flip();
byte[] bytes = new byte[buffer.remaining()];
buffer.get(bytes);
String body;
try {
body = new String(bytes, "UTF-8");
System.out.println("Now is : " + body);
latch.countDown();
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
// ingnore on close
}
}
});
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
// ingnore on close
}
}
});
}
@Override
public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
exc.printStackTrace();
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
}
package com.bj58.wuxian.aio;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {
private AsynchronousSocketChannel channel;
public ReadCompletionHandler(AsynchronousSocketChannel channel) {
if (this.channel == null)
this.channel = channel;
}
@Override
public void completed(Integer result, ByteBuffer attachment) {
attachment.flip();
byte[] body=new byte[attachment.remaining()];
attachment.get(body);
String req;
try {
req = new String(body,"utf-8");
System.out.println("The time server receive order : " + req);
String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req) ? new java.util.Date(
System.currentTimeMillis()).toString() : "BAD ORDER";
doWrite(currentTime);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
private void doWrite(String currentTime) {
if(currentTime!=null&& currentTime.trim().length()>0){
byte[] bytes=currentTime.getBytes();
ByteBuffer writeBuffer=ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
channel.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
// 如果沒有發(fā)送完成，繼續(xù)發(fā)送
if (attachment.hasRemaining())
channel.write(attachment, attachment, this);
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
channel.close();
} catch (IOException e) {
}
}
});
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
this.channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
我靠，是不是太麻煩了，用netty的話會(huì)簡(jiǎn)單很多，哈哈~~~~
