從JDK 1.4開始，Java的標(biāo)準(zhǔn)庫中就包含了NIO，
即所謂的“New
IO”。其中最重要的功能就是提供了“非阻塞”的IO，當(dāng)然包括了Socket。NonBlocking的IO就是對select(Unix平臺下)以及
WaitForMultipleObjects(Windows平臺)的封裝，提供了高性能、易伸縮的服務(wù)架構(gòu)。
說來慚愧，直到JDK1.4才有這種功能，但遲到者不一定沒有螃蟹吃，NIO就提供了優(yōu)秀的面向?qū)ο蟮慕鉀Q方案，可以很方便地編寫高性能的服務(wù)器。
話說回來，傳統(tǒng)的Server/Client實現(xiàn)是基于Thread per request，即服務(wù)器為每個客戶端請求建立一個線程處理，單獨負(fù)責(zé)處理一個客戶的請求。比如像Tomcat（新版本也會提供NIO方案）、Resin等Web服務(wù)器就是這樣實現(xiàn)的。當(dāng)然為了減少瞬間峰值問題，服務(wù)器一般都使用線程池，規(guī)定了同時并發(fā)的最大數(shù)量，避免了線程的無限增長。
但這樣有一個問題：如果線程池的大小為100，當(dāng)有100個用戶同時通過HTTP現(xiàn)在一個大文件時，服務(wù)器的線程池會用完，因為所有的線程都在傳輸大文件了，即使第101個請求者僅僅請求一個只有10字節(jié)的頁面，服務(wù)器也無法響應(yīng)了，只有等到線程池中有空閑的線程出現(xiàn)。
另外，線程的開銷也是很大的，特別是達到了一個臨界值后，性能會顯著下降，這也限制了傳統(tǒng)的Socket方案無法應(yīng)對并發(fā)量大的場合，而“非阻塞”的IO就能輕松解決這個問題。
下面只是一個簡單的例子：服務(wù)器提供了下載大型文件的功能，客戶端連接上服務(wù)器的12345端口后，就可以讀取服務(wù)器發(fā)送的文件內(nèi)容信息了。注意這里的服務(wù)器只有一個主線程，沒有其他任何派生線程，讓我們看看NIO是如何用一個線程處理N個請求的。
NIO服務(wù)器最核心的一點就是反應(yīng)器模式：當(dāng)有感興趣的事件發(fā)生的，就通知對
應(yīng)的事件處理器去處理這個事件，如果沒有，則不處理。所以使用一個線程做輪詢就可以了。當(dāng)然這里這是個例子，如果要獲得更高性能，可以使用少量的線程，一
個負(fù)責(zé)接收請求，其他的負(fù)責(zé)處理請求，特別是對于多CPU時效率會更高。
關(guān)于使用NIO過程中出現(xiàn)的問題，最為普遍的就是為什么沒有請求時CPU的占
用率為100%？出現(xiàn)這種問題的主要原因是注冊了不感興趣的事件，比如如果沒有數(shù)據(jù)要發(fā)到客戶端，而又注冊了寫事件（OP_WRITE），則在
Selector.select()上就會始終有事件出現(xiàn)，CPU就一直處理了，而此時select()應(yīng)該是阻塞的。
另外一個值得注意的問題是：由于只使用了一個線程（多個線程也如此）處理用戶請求，所以要避免線程被阻塞，解決方法是事件的處理者必須要即刻返回，不能陷入循環(huán)中，否則會影響其他用戶的請求速度。
具體到本例子中，由于文件比較大，如果一次性發(fā)送整個文件（這里的一次性不是
指send整個文件內(nèi)容，而是通過while循環(huán)不間斷的發(fā)送分組包），則主線程就會阻塞，其他用戶就不能響應(yīng)了。這里的解決方法是當(dāng)有WRITE事件
時，僅僅是發(fā)送一個塊（比如4K字節(jié)）。發(fā)完后，繼續(xù)等待WRITE事件出現(xiàn)，依次處理，直到整個文件發(fā)送完畢，這樣就不會阻塞其他用戶了。
服務(wù)器的例子：
package
nio.file;
import
java.io.FileInputStream;
import
java.io.IOException;
import
java.net.InetSocketAddress;
import
java.nio.ByteBuffer;
import
java.nio.CharBuffer;
import
java.nio.channels.FileChannel;
import
java.nio.channels.Selecti;
import
java.nio.channels.Selector;
import
java.nio.channels.ServerSocketChannel;
import
java.nio.channels.SocketChannel;
import
java.nio.charset.Charset;
import
java.nio.charset.CharsetDecoder;
import
java.util.Iterator;
/**
* 測試文件下載的NIOServer
*
*
@author
tenyears.cn
*/
public class
NIOServer
{
static
int
BLOCK =
4096
;
// 處理與客戶端的交互
public class
HandleClient
{
protected
FileChannel channel;
protected
ByteBuffer buffer;
public
HandleClient
()
throws
IOException
{
this
.channel =
new
FileInputStream
(
filename
)
.getChannel
()
;
this
.buffer = ByteBuffer.allocate
(
BLOCK
)
;
}
public
ByteBuffer readBlock
() {
try
{
buffer.clear
()
;
int
count = channel.read
(
buffer
)
;
buffer.flip
()
;
if
(
count <=
0
)
return null
;
}
catch
(
IOException e
) {
e.printStackTrace
()
;
}
return
buffer;
}
public
void
close
() {
try
{
channel.close
()
;
}
catch
(
IOException e
) {
e.printStackTrace
()
;
}
}
}
protected
Selector selector;
protected
String filename =
"d:\\bigfile.dat"
;
// a big file
protected
ByteBuffer clientBuffer = ByteBuffer.allocate
(
BLOCK
)
;
protected
CharsetDecoder decoder;
public
NIOServer
(
int
port
)
throws
IOException
{
selector =
this
.getSelector
(
port
)
;
Charset charset = Charset.forName
(
"GB2312"
)
;
decoder = charset.newDecoder
()
;
}
// 獲取Selector
protected
Selector getSelector
(
int
port
)
throws
IOException
{
ServerSocketChannel server = ServerSocketChannel.open
()
;
Selector sel = Selector.open
()
;
server.socket
()
.bind
(
new
InetSocketAddress
(
port
))
;
server.configureBlocking
(
false
)
;
server.register
(
sel, Selecti.OP_ACCEPT
)
;
return
sel;
}
// 監(jiān)聽端口
public
void
listen
() {
try
{
for
(
;;
) {
selector.select
()
;
Iterator iter = selector.selectedKeys
()
.iterator
()
;
while
(
iter.hasNext
()) {
Selecti key = iter.next
()
;
iter.remove
()
;
handleKey
(
key
)
;
}
}
}
catch
(
IOException e
) {
e.printStackTrace
()
;
}
}
// 處理事件
protected
void
handleKey
(
Selecti key
)
throws
IOException
{
if
(
key.isAcceptable
()) {
// 接收請求
ServerSocketChannel server =
(
ServerSocketChannel
)
key.channel
()
;
SocketChannel channel = server.accept
()
;
channel.configureBlocking
(
false
)
;
channel.register
(
selector, Selecti.OP_READ
)
;
}
else if
(
key.isReadable
()) {
// 讀信息
SocketChannel channel =
(
SocketChannel
)
key.channel
()
;
int
count = channel.read
(
clientBuffer
)
;
if
(
count >
0
) {
clientBuffer.flip
()
;
CharBuffer charBuffer = decoder.decode
(
clientBuffer
)
;
System.out.println
(
"Client >>"
+ charBuffer.toString
())
;
Selecti wKey = channel.register
(
selector,
Selecti.OP_WRITE
)
;
wKey.attach
(
new
HandleClient
())
;
}
else
channel.close
()
;
clientBuffer.clear
()
;
}
else if
(
key.isWritable
()) {
// 寫事件
SocketChannel channel =
(
SocketChannel
)
key.channel
()
;
HandleClient handle =
(
HandleClient
)
key.attachment
()
;
ByteBuffer block = handle.readBlock
()
;
if
(
block !=
null
)
channel.write
(
block
)
;
else
{
handle.close
()
;
channel.close
()
;
}
}
}
public static
void
main
(
String
[]
args
) {
int
port =
12345
;
try
{
NIOServer server =
new
NIOServer
(
port
)
;
System.out.println
(
"Listernint on "
+ port
)
;
while
(
true
) {
server.listen
()
;
}
}
catch
(
IOException e
) {
e.printStackTrace
()
;
}
}
}
|
該代碼中，通過一個HandleClient來獲取文件的一塊數(shù)據(jù)，每一個客戶都會分配一個HandleClient的實例。
下面是客戶端請求的代碼，也比較簡單，模擬100個用戶同時下載文件。
package
nio.file;
import
java.io.IOException;
import
java.net.InetSocketAddress;
import
java.nio.ByteBuffer;
import
java.nio.CharBuffer;
import
java.nio.channels.Selecti;
import
java.nio.channels.Selector;
import
java.nio.channels.SocketChannel;
import
java.nio.charset.Charset;
import
java.nio.charset.CharsetEncoder;
import
java.util.Iterator;
import
java.util.concurrent.ExecutorService;
import
java.util.concurrent.Executors;
/**
* 文件下載客戶端
*
@author
tenyears.cn
*/
public class
NIOClient
{
static
int
SIZE =
100
;
static
InetSocketAddress ip =
new
InetSocketAddress
(
"localhost"
,
12345
)
;
static
CharsetEncoder encoder = Charset.forName
(
"GB2312"
)
.newEncoder
()
;
static class
Download
implements
Runnable
{
protected
int
index;
public
Download
(
int
index
) {
this
.index = index;
}
public
void
run
() {
try
{
long
start = System.currentTimeMillis
()
;
SocketChannel client = SocketChannel.open
()
;
client.configureBlocking
(
false
)
;
Selector selector = Selector.open
()
;
client.register
(
selector, Selecti.OP_CONNECT
)
;
client.connect
(
ip
)
;
ByteBuffer buffer = ByteBuffer.allocate
(
8
*
1024
)
;
int
total =
0
;
FOR:
for
(
;;
) {
selector.select
()
;
Iterator iter = selector.selectedKeys
()
.iterator
()
;
while
(
iter.hasNext
()) {
Selecti key = iter.next
()
;
iter.remove
()
;
if
(
key.isConnectable
()) {
SocketChannel channel =
(
SocketChannel
)
key
.channel
()
;
if
(
channel.isConnectionPending
())
channel.finishConnect
()
;
channel.write
(
encoder.encode
(
CharBuffer
.wrap
(
"Hello from "
+ index
)))
;
channel.register
(
selector, Selecti.OP_READ
)
;
}
else if
(
key.isReadable
()) {
SocketChannel channel =
(
SocketChannel
)
key
.channel
()
;
int
count = channel.read
(
buffer
)
;
if
(
count >
0
) {
total += count;
buffer.clear
()
;
}
else
{
client.close
()
;
break
FOR;
}
}
}
}
double
last =
(
System.currentTimeMillis
()
- start
)
*
1.0
/
1000
;
System.out.println
(
"Thread "
+ index +
" downloaded "
+ total
+
"bytes in "
+ last +
"s."
)
;
}
catch
(
IOException e
) {
e.printStackTrace
()
;
}
}
}
public static
void
main
(
String
[]
args
)
throws
IOException
{
ExecutorService exec = Executors.newFixedThreadPool
(
SIZE
)
;
for
(
int
index =
0
; index < SIZE; index++
) {
exec.execute
(
new
Download
(
index
))
;
}
exec.shutdown
()
;
}
}
posted on 2007-01-19 00:03
苦笑枯 閱讀(473)
評論(0) 編輯 收藏 所屬分類:
Java