## 創建阻塞的伺服器 當 `ServerSocketChannel` 與 `SockelChannel` 採用預設的阻塞模式時,為了同時處理多個客戶的連接,必須使用多線程 ```java public class EchoServer { private int port = 8000; priv ...
創建阻塞的伺服器
當 ServerSocketChannel
與 SockelChannel
採用預設的阻塞模式時,為了同時處理多個客戶的連接,必須使用多線程
public class EchoServer {
private int port = 8000;
private ServerSocketChannel serverSocketChannel = null;
private ExecutorService executorService; //線程池
private static final int POOL_MULTIPLE = 4; //線程池中工作線程的數目
public EchoServer() throws IOException {
//創建一個線程池
executorService = Executors.newFixedThreadPool(
Runtime.getRuntime().availableProcessors() * POOL_MULTIPLE);
//創建一個ServerSocketChannel對象
serverSocketChannel = ServerSocketChannel.open();
//使得在同一個主機上關閉了伺服器程式,緊接著再啟動該伺服器程式時,可以順利綁定相同的埠
serverSocketChannel.socket().setReuseAddress(true);
//把伺服器進程與一個本地埠綁定
serverSocketChannel.socket().bind(new InetSocketAddress(port));
System.out.println("伺服器啟動");
}
public void service() {
while (true) {
SocketChannel socketChannel = null;
try {
socketChannel = serverSocketChannel.accept();
//處理客戶連接
executorService.execute(new Handler(socketChannel));
} catch(IOException e) {
e.printStackTrace();
}
}
}
public static void main(String args[])throws IOException {
new EchoServer().service();
}
//處理客戶連按
class Handler implements Runnable {
private SocketChannel socketChannel;
public Handler(SocketChannel socketChannel) {
this.socketChannel = socketChannel;
}
public void run() {
handle(socketChannel);
}
public void handle(SocketChannel socketChannel) {
try {
//獲得與socketChannel關聯的Socket對象
Socket socket = socketChannel.socket();
System.out.println("接收到客戶連接,來自:" + socket.getInetAddress() + ":" + socket.getPort());
BufferedReader br = getReader(socket);
PrintWriter pw = getWriter(socket);
String msg = null;
while ((msg = br.readLine()) != null) {
System.out.println(msg);
pw.println(echo(msg));
if (msg.equals("bye")) {
break;
}
}
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
if(socketChannel != null) {
socketChannel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
private PrintWriter getWriter(Socket socket) throws IOException {
OutputStream socketOut = socket.getOutputStream();
return new PrintWriter(socketOut,true);
}
private BufferedReader getReader(Socket socket) throws IOException {
InputStream socketIn = socket.getInputStream();
return new BufferedReader(new InputStreamReader(socketIn));
}
public String echo(String msg) {
return "echo:" + msg;
}
}
創建非阻塞的伺服器
在非阻塞模式下,EchoServer
只需要啟動一個主線程,就能同時處理三件事:
- 接收客戶的連接
- 接收客戶發送的數據
- 向客戶發迴響應數據
EchoServer
委托 Selector
來負責監控接收連接就緒事件、讀就緒事件和寫就緒事件如果有特定事件發生,就處理該事件
// 創建一個Selector對象
selector = Selector.open();
//創建一個ServerSocketChannel對象
serverSocketChannel = ServerSocketChannel.open();
//使得在同一個主機上關閉了伺服器程式,緊接著再啟動該伺服器程式時
//可以順利綁定到相同的埠
serverSocketChannel.socket().setReuseAddress(true);
//使ServerSocketChannel工作於非阻塞模式
serverSocketChannel.configureBlocking(false):
//把伺服器進程與一個本地埠綁定
serverSocketChannelsocket().bind(new InetSocketAddress(port));
EchoServer
類的 service()
方法負責處理本節開頭所說的三件事,體現其主要流程的代碼如下:
public void service() throws IOException {
serverSocketChannel.reqister(selector, SelectionKey.OP_ACCEPT);
//第1層while迴圈
while(selector.select() > 0) {
//獲得Selector的selected-keys集合
Set readyKeys = selector.selectedKeys();
Iterator it = readyKeys.iterator();
//第2層while迴圈
while (it.hasNext()) {
SelectionKey key = null;
//處理SelectionKey
try {
//取出一個SelectionKey
key = (SelectionKey) it.next();
//把 SelectionKey從Selector 的selected-key 集合中刪除
it.remove();
1f (key.isAcceptable()) { 處理接收連接就緒事件; }
if (key.isReadable()) { 處理讀就緒水件; }
if (key.isWritable()) { 處理寫就緒事件; }
} catch(IOException e) {
e.printStackTrace();
try {
if(key != null) {
//使這個SelectionKey失效
key.cancel();
//關閉與這個SelectionKey關聯的SocketChannel
key.channel().close();
}
} catch(Exception ex) {
e.printStackTrace();
}
}
}
}
}
- 首先由
ServerSocketChannel
向Selector
註冊接收連接就緒事件,如果Selector
監控到該事件發生,就會把相應的SelectionKey
對象加入selected-keys
集合 - 第一層 while 迴圈,不斷詢問
Selector
已經發生的事件,select()
方法返回當前相關事件已經發生的SelectionKey
的個數,如果當前沒有任何事件發生,該方法會阻塞下去,直到至少有一個事件發生。Selector
的selectedKeys()
方法返回selected-keys
集合,它存放了相關事件已經發生的SelectionKey
對象 - 第二層 while 迴圈,從
selected-keys
集合中依次取出每個SelectionKey
對象並從集合中刪除,,然後調用isAcceptable()
、isReadable()
和isWritable()
方法判斷到底是哪種事件發生了,從而做出相應的處理
1. 處理接收連接就緒事件
if (key.isAcceptable()) {
//獲得與SelectionKey關聯的ServerSocketChannel
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
//獲得與客戶連接的SocketChannel
SocketChannel socketChannel = (SocketChannel) ssc.accept();
//把Socketchannel設置為非阻塞模式
socketChannel.configureBlocking(false);
//創建一個用於存放用戶發送來的數據的級沖區
ByteBuffer buffer = ByteBuffer.allocate(1024);
//Socketchannel向Selector註冊讀就緒事件和寫就緒事件
socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
}
2. 處理讀就緒事件
public void receive(SelectionKey key) throws IOException {
//獲得與SelectionKey關聯的附件
ByteBuffer buffer = (ByteBuffer) key.attachment();
//獲得與SelectionKey關聯的Socketchannel
SocketChannel socketChannel = (SocketChannel)key.channel();
//創建一個ByteBuffer用於存放讀到的數據
ByteBuffer readBuff = ByteBuffer.allocate(32);
socketChannel.read(readBuff);
readBuff.flip();
//把buffer的極限設為容量
buffer.limit(buffer.capacity());
//把readBuff中的內容拷貝到buffer
buffer.put(readBuff);
}
3. 處理寫就緒事件
public void send(SelectionKey key) throws IOException {
//獲得與SelectionKey關聯的ByteBuffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
//獲得與SelectionKey關聯的SocketChannel
SocketChannel socketChannel = (SocketChannel) key.channel();
buffer.flip();
//按照GBK編碼把buffer中的位元組轉換為字元串
String data = decode(buffer);
//如果還沒有讀到一行數據就返回
if(data.indexOf("\r\n") == -1)
return;
//截取一行數據
String outputData = data.substring(0, data.indexOf("\n") + 1);
//把輸出的字元串按照GBK編碼轉換為位元組,把它放在outputBuffer中
ByteBuffer outputBuffer = encode("echo:" + outputData);
//輸出outputBuffer的所有位元組
while(outputBuffer,hasRemaining())
socketChannel.write(outputBuffer);
//把outputData字元審按照GBK編碼,轉換為位元組,把它放在ByteBuffer
ByteBuffer temp = encode(outputData);
//把buffer的位置設為temp的極限
buffer.position(temp.limit()):
//刪除buffer已經處理的數據
buffer.compact();
//如果已經輸出了字元串“bye\r\n”,就使SelectionKey失效,並關閉SocketChannel
if(outputData.equals("bye\r\n")) {
key.cancel();
socketChannel.close();
}
}
完整代碼如下:
public class EchoServer {
private int port = 8000;
private ServerSocketChannel serverSocketChannel = null;
private Selector selector;
private Charset charset = Charset.forName("GBK");
public EchoServer() throws IOException {
// 創建一個Selector對象
selector = Selector.open();
//創建一個ServerSocketChannel對象
serverSocketChannel = ServerSocketChannel.open();
//使得在同一個主機上關閉了伺服器程式,緊接著再啟動該伺服器程式時
//可以順利綁定到相同的埠
serverSocketChannel.socket().setReuseAddress(true);
//使ServerSocketChannel工作於非阻塞模式
serverSocketChannel.configureBlocking(false):
//把伺服器進程與一個本地埠綁定
serverSocketChannelsocket().bind(new InetSocketAddress(port));
}
public void service() throws IOException {
serverSocketChannel.reqister(selector, SelectionKey.OP_ACCEPT);
//第1層while迴圈
while(selector.select() > 0) {
//獲得Selector的selected-keys集合
Set readyKeys = selector.selectedKeys();
Iterator it = readyKeys.iterator();
//第2層while迴圈
while (it.hasNext()) {
SelectionKey key = null;
//處理SelectionKey
try {
//取出一個SelectionKey
key = (SelectionKey) it.next();
//把 SelectionKey從Selector 的selected-key 集合中刪除
it.remove();
1f (key.isAcceptable()) {
//獲得與SelectionKey關聯的ServerSocketChannel
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
//獲得與客戶連接的SocketChannel
SocketChannel socketChannel = (SocketChannel) ssc.accept();
//把Socketchannel設置為非阻塞模式
socketChannel.configureBlocking(false);
//創建一個用於存放用戶發送來的數據的級沖區
ByteBuffer buffer = ByteBuffer.allocate(1024);
//Socketchannel向Selector註冊讀就緒事件和寫就緒事件
socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
}
if (key.isReadable()) { receive(key); }
if (key.isWritable()) { send(key); }
} catch(IOException e) {
e.printStackTrace();
try {
if(key != null) {
//使這個SelectionKey失效
key.cancel();
//關閉與這個SelectionKey關聯的SocketChannel
key.channel().close();
}
} catch(Exception ex) {
e.printStackTrace();
}
}
}
}
}
public void receive(SelectionKey key) throws IOException {
//獲得與SelectionKey關聯的附件
ByteBuffer buffer = (ByteBuffer) key.attachment();
//獲得與SelectionKey關聯的Socketchannel
SocketChannel socketChannel = (SocketChannel)key.channel();
//創建一個ByteBuffer用於存放讀到的數據
ByteBuffer readBuff = ByteBuffer.allocate(32);
socketChannel.read(readBuff);
readBuff.flip();
//把buffer的極限設為容量
buffer.limit(buffer.capacity());
//把readBuff中的內容拷貝到buffer
buffer.put(readBuff);
}
public void send(SelectionKey key) throws IOException {
//獲得與SelectionKey關聯的ByteBuffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
//獲得與SelectionKey關聯的SocketChannel
SocketChannel socketChannel = (SocketChannel) key.channel();
buffer.flip();
//按照GBK編碼把buffer中的位元組轉換為字元串
String data = decode(buffer);
//如果還沒有讀到一行數據就返回
if(data.indexOf("\r\n") == -1)
return;
//截取一行數據
String outputData = data.substring(0, data.indexOf("\n") + 1);
//把輸出的字元串按照GBK編碼轉換為位元組,把它放在outputBuffer中
ByteBuffer outputBuffer = encode("echo:" + outputData);
//輸出outputBuffer的所有位元組
while(outputBuffer,hasRemaining())
socketChannel.write(outputBuffer);
//把outputData字元審按照GBK編碼,轉換為位元組,把它放在ByteBuffer
ByteBuffer temp = encode(outputData);
//把buffer的位置設為temp的極限
buffer.position(temp.limit()):
//刪除buffer已經處理的數據
buffer.compact();
//如果已經輸出了字元串“bye\r\n”,就使SelectionKey失效,並關閉SocketChannel
if(outputData.equals("bye\r\n")) {
key.cancel();
socketChannel.close();
}
}
//解碼
public String decode(ByteBuffer buffer) {
CharBuffer charBuffer = charset.decode(buffer);
return charBuffer.toStrinq();
}
//編碼
public ByteBuffer encode(String str) {
return charset.encode(str);
}
public static void main(String args[])throws Exception {
EchoServer server = new EchoServer();
server.service();
}
}
阻塞模式與非阻塞模式混合使用
使用非阻塞模式時,ServerSocketChannel
以及 SocketChannel
都被設置為非阻塞模式,這使得接收連接、接收數據和發送數據的操作都採用非阻塞模式,EchoServer
採用一個線程同時完成這些操作
假如有許多客戶請求連接,可以把接收客戶連接的操作單獨由一個線程完成,把接收數據和發送數據的操作由另一個線程完成,這可以提高伺服器的併發性能
負責接收客戶連接的線程按照阻塞模式工作,如果收到客戶連接,就向 Selector
註冊讀就緒和寫就緒事件,否則進入阻塞狀態,直到接收到了客戶的連接。負責接收數據和發送數據的線程按照非阻塞模式工作,只有在讀就緒或寫就緒事件發生時,才執行相應的接收數據和發送數據操作
public class EchoServer {
private int port = 8000;
private ServerSocketChannel serverSocketChannel = null;
private Selector selector = null;
private Charset charset = Charset.forName("GBK");
public EchoServer() throws IOException {
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().setReuseAddress(true);
serverSocketChannelsocket().bind(new InetSocketAddress(port));
}
public void accept() {
while(true) {
try {
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(1024);
synchronized(gate) {
selector.wakeup();
socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
}
} catch(IOException e) {
e.printStackTrace();
}
}
}
private Object gate=new Object();
public void service() throws IOException {
while(true) {
synchronized(gate){}
int n = selector.select();
if(n == 0) continue;
Set readyKeys = selector.selectedKeys();
Iterator it = readyKeys.iterator();
while (it.hasNext()) {
SelectionKey key = null;
try {
it.remove();
if (key.isReadable()) {
receive(key);
}
if (key.isWritable()) {
send(key);
}
} catch(IOException e) {
e.printStackTrace();
try {
if(key != null) {
key.cancel();
key.channel().close();
}
} catch(Exception ex) { e.printStackTrace(); }
}
}
}
}
public void receive(SelectionKey key) throws IOException {
...
}
public void send(SelectionKey key) throws IOException {
...
}
public String decode(ByteBuffer buffer) {
...
}
public ByteBuffer encode(String str) {
...
}
public static void main(String args[])throws Exception {
final EchoServer server = new EchoServer();
Thread accept = new Thread() {
public void run() {
server.accept();
}
};
accept.start();
server.service();
}
}
註意一點:主線程的 selector select()
方法和 Accept 線程的 register(...)
方法都會造成阻塞,因為他們都會操作 Selector
對象的共用資源 all-keys
集合,這有可能會導致死鎖
導致死鎖的具體情形是:Selector
中尚沒有任何註冊的事件,即 all-keys
集合為空,主線程執行 selector.select()
方法時將進入阻塞狀態,只有當 Accept 線程向 Selector
註冊了事件,並且該事件發生後,主線程才會從 selector.select()
方法返回。然而,由於主線程正在 selector.select()
方法中阻塞,這使得 Acccept
線程也在 register()
方法中阻塞。Accept 線程無法向 Selector 註冊事件,而主線程沒有任何事件可以監控,所以這兩個線程將永遠阻塞下去
為了避免對共用資源的競爭,同步機制使得一個線程執行 register()
時,不允許另一個線程同時執行 select()
方法,反之亦然