基於執行緒池和NIO技術構建高效的多協議Android通訊框架

作者孫東風 2011-1-20轉載請註明出處

引言

在多數涉及網路通訊的手機應用中，由於GPRS網路的速度在目前的情況下還不算理想，所以，如何能夠高效的請求得到網路資料就成為大多數應用所面臨的瓶頸問題。同時，在一些應用程式中可能會使用多種協議，比如IM通訊、視訊流型別的應用會犧牲資料的完整性來更高效的獲取資料，在這種型別的應用中，可能需要同時支援TCP、UDP以及HTTP協議。本文就嘗試基於Android的多執行緒技術ThreadPoolExecutor以及NIO非阻塞式程式設計構建這樣一個框架，以高效的獲取網路資料並很好的支援多種協議的併發請求。

基本設計思路

既然是基於ThreadPoolExecutor執行緒池來管理多個NIO執行緒的請求的，那麼首先應該有個全域性的ThreadPoolExecutor變數，使用單例模式來實現：

public static synchronized ThreadPoolExecutor setThreadPoolNum(int aThreadPoolMinNum,

int aThreadPoolMaxNum,long keepAliveTime)

{

if(threadPool == null)

{

threadPool =

new ThreadPoolExecutor(aThreadPoolMinNum,aThreadPoolMaxNum,

keepAliveTime,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(3),

new ThreadPoolExecutor.DiscardOldestPolicy());

}

return threadPool;

}

每個NIO請求被設計為單獨的執行緒，而每個網路通訊的連線都可以附屬在NIO執行緒上，基本的設計思路如圖1所示：

 

圖1

NIO通訊執行緒的設計

NIO通訊執行緒被設計為管理多個不同型別的網路連線，並負責維護每個連線所對應的網路資料處理介面，這些網路資料處理介面包括資料傳送介面processWrite()

、資料接收介面processRead()以及錯誤處理介面processError()等。

在每個NIO通訊執行緒被執行緒池物件啟動之後，首先檢查NIO埠是否有連線註冊的資料到來，如果有資料到來，則提交給相應的連線進行處理，程式碼如下：

try {

if(selector != null)

n = selector.select(3000);

// 如果要shutdown，關閉selector退出

if (shutdown) {

selector.close();

break;

}

} catch (IOException e) {

dispatchErrorToAll(e);

}

// 如果select返回大於0，處理事件

if(n > 0) {

for (Iterator<SelectionKey> i = selector.selectedKeys().iterator(); i.hasNext();) {

// 得到下一個Key

SelectionKey sk = i.next();

i.remove();

// 檢查其是否還有效

if(!sk.isValid())

continue;

// 處理

INIOHandler handler = (INIOHandler)sk.attachment();

try {

if(sk.isConnectable())

{

handler.processConnect(sk);

}

else if (sk.isReadable())

{

handler.processRead(sk);

}

} catch (IOException e) {

handler.processError(e);

} catch (RuntimeException e) {

}

}

n = 0;

}

checkNewConnection();

notifySend();

上面的程式碼首先使用Selector類的selectedKyes()方法獲取到所有有事件發生的連線通道，然後遍歷這些通道，並使用attachment()方法獲取到註冊到通道上的處理物件NIOHandler，之後呼叫每個連線通道上註冊的NIOHandler物件的方法進行網路資料的處理。

網路連線的設計

從NIO通訊執行緒的設計中知道，每個網路連線都有相應的連線通道以及連線通道的資料處理器，那麼就可以抽象出來這些連線通道以及資料處理器的介面：

public interface IConnection {

/**

* 新增一個包到傳送佇列

*

* @param out

* OutPacket子類

*/

public void add(OutPacket out);

public void clearSendQueue();

public void start();

public String getId();

public void dispose();

public InetSocketAddress getRemoteAddress();

public SelectableChannel channel();

public INIOHandler getNIOHandler();

public boolean isEmpty();

public void receive() throws IOException;

public void send() throws IOException;

public void send(ByteBuffer buffer);

public boolean isConnected();

}

在通道資料處理器中，每個連線通道都應該有基本的資料傳送、資料接收以及錯誤處理介面，可以抽象出資料處理器的基本介面如下：

public interface INIOHandler {

public void processConnect(SelectionKey sk) throws IOException;

public void processRead(SelectionKey sk) throws IOException;

public void processWrite() throws IOException;

public void processError(Exception e);

}

那麼每個網路連線都應該繼承上述兩個介面，比如TCP連線需要有SocketChannel通道以及相應的NIOHandler資料處理介面，UDP連線需要有DatagramChannel通道以及相應的NIOHandler資料處理介面，下面是TCPConnection的程式碼：

public class TCPConnection extends ConnectionImp{

/** 用於通訊的channel */

private final SocketChannel channel;

/**

* true表示遠端已經關閉了這個連線

*/

private boolean remoteClosed;

/**

* 構造一個連線到指定地址的TCPPort.

*

* @param address 連線到的地址.

* @throws IOException 埠開啟/埠配置/連線到地址出錯.

*/

public TCPConnection(String id, InetSocketAddress address) throws IOException {

super(id);

channel = SocketChannel.open();

channel.configureBlocking(false);

this.remoteAddress = address;

remoteClosed = false;

}

public void start() {

try {

channel.connect(remoteAddress);

} catch(UnknownHostException e) {

processError(new Exception("Unknown Host"));

} catch(UnresolvedAddressException e) {

processError(new Exception("Unable to resolve server address"));

} catch (IOException e) {

processError(e);

}

}

public SelectableChannel channel() {

return channel;

}

public void receive() throws IOException {

if(remoteClosed)

return;

//接收資料

int oldPos = receiveBuf.position();

for (int r = channel.read(receiveBuf); r > 0; r = channel.read(receiveBuf))

;

byte[] tempBuffer = new byte[1024];

receiveBuf.get(tempBuffer, 0, receiveBuf.position());

Log.e("receive", " = "+new String(tempBuffer,"UTF-8"));

// 得到當前位置

int pos = receiveBuf.position();

receiveBuf.flip();

// 檢查是否讀了0位元組，這種情況一般表示遠端已經關閉了這個連線

if(oldPos == pos) {

remoteClosed = true;

return;

}

InPacket packet = new InPacket(receiveBuf);

inQueue.add(packet);

adjustBuffer(pos);

}

private void adjustBuffer(int pos) {

// 如果0不等於當前pos，說明至少分析了一個包

if(receiveBuf.position() > 0) {

receiveBuf.compact();

receiveBuf.limit(receiveBuf.capacity());

} else {

receiveBuf.limit(receiveBuf.capacity());

receiveBuf.position(pos);

}

}

public void send() throws IOException {

while (!isEmpty()) {

sendBuf.clear();

OutPacket packet = remove();

channel.write(ByteBuffer.wrap(packet.getBody()));

// 新增到重發佇列

packet.setTimeout(System.currentTimeMillis() + EnginConst.QQ_TIMEOUT_SEND);

Log.e("debug","have sended packet - " + packet.toString());

}

}

public void send(OutPacket packet) {

try {

sendBuf.clear();

channel.write(ByteBuffer.wrap(packet.getBody()));

Log.d("debug","have sended packet - " + packet.toString());

} catch (Exception e) {

}

}

public void send(ByteBuffer buffer) {

try {

channel.write(buffer);

} catch (IOException e) {

}

}

public void dispose() {

try {

channel.close();

} catch(IOException e) {

}

}

public boolean isConnected() {

return channel != null && channel.isConnected();

}

public void processConnect(SelectionKey sk) throws IOException {

//完成SocketChannel的連線

channel.finishConnect();

while(!channel.isConnected()) {

try {

Thread.sleep(300);

} catch (InterruptedException e) {

}

channel.finishConnect();

}

sk.interestOps(SelectionKey.OP_READ);

Log.e("debug","hava connected to server");

}

public void processRead(SelectionKey sk) throws IOException {

receive();

}

public void processWrite() throws IOException {

if(isConnected())

send();

}

}

測試程式碼

public void onCreate(Bundle savedInstanceState) {

super.onCreate(savedInstanceState);

setContentView(R.layout.main);

AsyncRequest request = new AsyncRequest(new InetSocketAddress("10.0.2.2",8042));

try {

request.setConNumPerNIOThread(5);

AsyncRequest.setThreadPoolNum(2, 5, 2);

request.setUDP(false);

request.setBody("hello world".getBytes());

request.startAsyn(0);

if(!request.isCurrRequestFull())

{

request.setBody("My name is dongfengsun".getBytes());

request.startAsyn(1);

}

} catch (Exception e) {

Log.e("onCreate", " = "+e.getMessage());

e.printStackTrace();

}

}

測試程式碼中設定每個NIO通訊執行緒的連線上限為5，如果每個NIO通訊執行緒還有空餘的連線數，那麼可以繼續在當前的NIO通訊執行緒建立網路連線，如果滿了，則可以重新初始化一個NIO通訊執行緒並加入到執行緒池中，上面的程式碼在伺服器端控制檯列印的資料如圖2所示：

圖2

因為CSDN上無法上傳原始碼，所以需要原始碼的可以在回覆中留下信箱，有時間我會統一發送^_^

原始碼不再逐個傳送，已上傳到口袋網 http://www.pocketcn.com上，請到網站Android板塊進行下載