|
對于用ServerSocket和Socket寫的服務器程序或著客戶端程序，在運行的時候常常會阻塞，如當一個線程執(zhí)行ServerSocket的accept()方法，如果沒有客戶機連接，該線程就會一直阻塞直到有了客戶機連接才從accept()方法返回，再如，當線程執(zhí)行Socket的read()方法，如果輸入流中沒有數(shù)據(jù)，該線程就會一直等到有數(shù)據(jù)可讀時才從read()方法返回。
如果服務器要與多個客戶機通信，通常做法為每個客戶機連接開啟一個服務線程，每個工作線程都有可能經(jīng)常處于長時間的阻塞狀態(tài)。
從JDK1.4版本開始，引入了非阻塞的通信機制。服務器程序接收客戶連接、客戶程序建立與服務器的連接，以及服務器程序和客戶程序收發(fā)數(shù)據(jù)的操作都可以按非阻塞的方式進行。服務器程序只需要創(chuàng)建一個線程，就能完成同時與多個客戶通信的任務。
非阻塞通信要比傳統(tǒng)的阻塞方式效率要高，Apache的MIMA框架就是以java.nio包中的類編寫的。
不知道是否有朋友看過 孫衛(wèi)琴寫的《Java網(wǎng)絡編程精解》，在提到線程阻塞的時
Java網(wǎng)絡編程精解 第84頁 寫道
線程從Socket的輸入流讀入數(shù)據(jù)時，如果沒有足夠的數(shù)據(jù)，就會進入阻塞狀態(tài)，直到讀到了足夠的數(shù)據(jù)，或者到達輸入流的未尾，或者出現(xiàn)了異常，才從輸入流的read()方法返回或異常中斷。輸入流中有多少數(shù)據(jù)才算足夠呢，這要看線程執(zhí)行read方法的類型。
int read()：只要輸入有一個字節(jié)，就算足夠。
int read(byte[] buff)：只要輸入流中的字節(jié)數(shù)目與參數(shù)buff數(shù)組的長度相同，就算足夠。
我對描紅的描述持不同的意見
byte[] msgBytes = new byte[512];
inputStream.read(msgBytes);
如果按書中描述，這行代碼必須讀到512個字節(jié)后才從阻塞狀態(tài)中返回，如果沒有讀到足夠的512個字節(jié)，則一直阻塞。但實際情況卻不是這樣的，只要流里哪怕只有一個字節(jié) ,inputStream.read(msgBytes)也會立即返回，返回值為讀到的字節(jié)數(shù)。
下面是簡單的NIO示例
1、服務端程序，非阻塞方式
package com.bill99.nioserver;
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.nio.channels.spi.SelectorProvider;
import java.nio.charset.Charset;
import java.util.Iterator;
public class NIOServer {
private Selector socketSelector = null;
private ServerSocketChannel ssChannel = null;
private SocketChannel socketChannel =null;
private static SelectionKey key = null;
private int port =5512;
private int backlog = 100;
private Charset charset = Charset.defaultCharset();
private ByteBuffer shareBuffer = ByteBuffer.allocate(1024);//生成1kb的緩沖區(qū),可以根據(jù)實際情況調的更大些
public NIOServer() {
try {
socketSelector = SelectorProvider.provider().openSelector();
ssChannel =ServerSocketChannel.open() ;
ssChannel.socket().bind(new InetSocketAddress(port),100);
System.out.println(String.format("NIO服務器啟動,監(jiān)聽端口%1$s,最大連接數(shù)%2$s", port,backlog));
} catch(IOException e){
throw new ExceptionInInitializerError(e);
}
}
/**
* 接收客戶端連接
*/
public void acceptConnect() {
while(true) {
try {
SocketChannel socketChannel = ssChannel.accept();//阻塞模式，直到有連接進入
System.out.println("收到客戶機連接，來自:"+ssChannel.socket().getInetAddress());
socketChannel.configureBlocking(false);//設置非阻塞模式
synchronized(this){
socketSelector.wakeup();
socketChannel.register(socketSelector,SelectionKey.OP_READ|
SelectionKey.OP_WRITE);
}
} catch(IOException e){e.printStackTrace();}
}
}
/**
* 讀寫服務
* @throws IOException
*/
public void service() throws IOException{
while (true) {
synchronized (this) {//空的同步塊，目的是為了避免死鎖
}
if (!(socketSelector.select() > 0)) {
continue;
}
Iterator<SelectionKey> it = socketSelector.selectedKeys().iterator();
while (it.hasNext()) {
key = it.next();
it.remove();
if(key.isReadable()) {// 讀就緒
this.readDataFromSocket(key);
}
if(key.isWritable()){//寫就緒
this.sayWelcome(key);
}
}
}
}
//讀取客戶機發(fā)來的數(shù)據(jù)
private void readDataFromSocket(SelectionKey key) throws IOException {
shareBuffer.clear();//清空buffer
socketChannel=(SocketChannel) key.channel();
int num=0;
while((num = socketChannel.read(shareBuffer))>0){
shareBuffer.flip();//將當前極限設置為位置，并把設置后的位置改為0
}
if(num ==-1){//讀取流的未尾，對方已關閉流
socketChannel.close();
return;
}
System.out.println("client request:"+charset.decode(shareBuffer).toString());
}
//向客戶機發(fā)響應信息
private void sayWelcome(SelectionKey key) throws IOException {
shareBuffer.clear();//清空buffer
socketChannel=(SocketChannel) key.channel();
shareBuffer.put("Welcome to china!this is a greate and very beautifual country!\n".getBytes());
shareBuffer.flip();//將當前極限設置為位置，并把設置后的位置改為0
socketChannel.write(shareBuffer);
}
//Main方法
public static void main(String[] args) {
final NIOServer server = new NIOServer();
Runnable task = new Runnable() {
public void run() {
server.acceptConnect();
}
};
new Thread(task).start();//啟動處理客戶機連接的線程
try {
server.service();
} catch (IOException e) {//發(fā)生IO流異常時，關閉對應的socket
try {
key.channel().close();
key.cancel();
} catch (IOException e1) {
e1.printStackTrace();
}
}
}
}
2、客戶端程序，阻塞方式
- package com.bill99.client;
-
- import java.io.IOException;
- import java.io.InputStream;
- import java.io.OutputStream;
- import java.net.Socket;
- import java.nio.CharBuffer;
-
- import javax.net.SocketFactory;
-
- public class BlockingClient {
- private Socket socket = null;
- private OutputStream out = null;
- private InputStream in = null;
-
- public BlockingClient() {
- try {
- socket= SocketFactory.getDefault().createSocket("127.0.0.1", 5512);
- out = socket.getOutputStream();
- in = socket.getInputStream();
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
-
- public String receiveRespMsg(String reqMsg) throws IOException{
- out.write(reqMsg.getBytes());
- out.flush();
- in = socket.getInputStream();
- int c =0;
- CharBuffer buffer = CharBuffer.allocate(1024);
- while((c=in.read())!=-1 && c!=10){
- buffer.put((char)c);
- }
- return new String(buffer.array()).trim();
- }
-
- public static void main(String[] args) throws Exception{
- BlockingClient client = new BlockingClient();
- System.out.println("服務器響應:"+client.receiveRespMsg("hello\n"));
- }
- }
package com.bill99.client;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.nio.CharBuffer;
import javax.net.SocketFactory;
//測試類
public class BlockingClient {
private Socket socket = null;
private OutputStream out = null;
private InputStream in = null;
public BlockingClient() {
try {
socket= SocketFactory.getDefault().createSocket("127.0.0.1", 5512);
out = socket.getOutputStream();
in = socket.getInputStream();
} catch (IOException e) {
e.printStackTrace();
}
}
//發(fā)送請求并接收應答
public String receiveRespMsg(String reqMsg) throws IOException{
out.write(reqMsg.getBytes());
out.flush();
in = socket.getInputStream();
int c =0;
CharBuffer buffer = CharBuffer.allocate(1024);
while((c=in.read())!=-1 && c!=10){
buffer.put((char)c);
}
return new String(buffer.array()).trim();
}
public static void main(String[] args) throws Exception{
BlockingClient client = new BlockingClient();
System.out.println("服務器響應:"+client.receiveRespMsg("hello\n"));
}
}
總體而信，阻塞模式和非阻塞模式都可以同時處理多個客戶機的連接，但阻塞模式需要較多的線程許多時間都浪費在阻塞I/O操作上，Java虛擬機需要頻繁地轉讓CPU的使用權，而非阻塞模式只需要少量線程即可完成所有任務，非阻塞模式能更有效的利用CPU，系統(tǒng)開銷小，能夠提高程序的并發(fā)性能。
|
