Android開發進階之NIO非阻塞包(八)

 在整個DDMS中體現android NIO主要框架的要數MonitorThread.Java這個文件了，有關PC和android手機同步以及NIO非阻塞編程的精髓可以在下面的文件中充分體現出來。

final class MonitorThread extends Thread {

private static final int CLIENT_READY = 2;

private static final int CLIENT_DISCONNECTED = 3;

private volatile boolean mQuit = false;

private ArrayList<Client> mClIEntList; //用一個數組保存客戶端信息

private Selector mSelector;

private HashMap<Integer, ChunkHandler> mHandlerMap; //這裡android123提示大家，由於在多線程中concurrentHashMap效率比HashMap更安全高效，推薦使用並發庫的這個替代版本。

private ServerSocketChannel mDebugSelectedChan; //一個用於調試的服務器通道

private int mNewDebugSelectedPort;

private int mDebugSelectedPort = -1;

private Client mSelectedClIEnt = null;

private static MonitorThread mInstance;

private MonitorThread() {
super("Monitor");
mClientList = new ArrayList<ClIEnt>();
mHandlerMap = new HashMap<Integer, ChunkHandler>();

mNewDebugSelectedPort = DdmPreferences.getSelectedDebugPort();
}

static MonitorThread createInstance() { //創建實例
return mInstance = new MonitorThread();
}

static MonitorThread getInstance() { //獲取實例
return mInstance;
}

synchronized void setDebugSelectedPort(int port) throws IllegalStateException { //設置調試端口號
if (mInstance == null) {
return;
}

if (androidDebugBridge.getClIEntSupport() == false) {
return;
}

if (mDebugSelectedChan != null) {
Log.d("ddms", "Changing debug-selected port to " + port);
mNewDebugSelectedPort = port;
wakeup(); //這裡用來喚醒所有的Selector
} else {
// we set mNewDebugSelectedPort instead of mDebugSelectedPort so that it's automatically
mNewDebugSelectedPort = port;
}
}

synchronized void setSelectedClient(Client selectedClIEnt) {
if (mInstance == null) {
return;
}

if (mSelectedClient != selectedClIEnt) {
Client oldClient = mSelectedClIEnt;
mSelectedClient = selectedClIEnt;

if (oldClIEnt != null) {
oldClient.update(ClIEnt.CHANGE_PORT);
}

if (mSelectedClIEnt != null) {
mSelectedClient.update(ClIEnt.CHANGE_PORT);
}
}
}

Client getSelectedClIEnt() {
return mSelectedClIEnt;
}

boolean getRetryOnBadHandshake() {
return true; // TODO? make configurable
}

Client[] getClIEnts() {
synchronized (mClIEntList) {
return mClientList.toArray(new ClIEnt[0]);
}
}

synchronized void registerChunkHandler(int type, ChunkHandler handler) {
if (mInstance == null) {
return;
}

synchronized (mHandlerMap) {
if (mHandlerMap.get(type) == null) {
mHandlerMap.put(type, handler);
}
}
}

@Override
public void run() { //本類的主要線程
Log.d("ddms", "Monitor is up");

try {
mSelector = Selector.open();
} catch (IOException ioe) {
Log.logAndDisplay(LogLevel.ERROR, "ddms",
"Failed to initialize Monitor Thread: " + ioe.getMessage());
return;
}

while (!mQuit) {

try {
synchronized (mClIEntList) {
}

try {
if (androidDebugBridge.getClIEntSupport()) {
if ((mDebugSelectedChan == null ||
mNewDebugSelectedPort != mDebugSelectedPort) &&
mNewDebugSelectedPort != -1) {
if (reopenDebugSelectedPort()) {
mDebugSelectedPort = mNewDebugSelectedPort;
}
}
}
} catch (IOException ioe) {
Log.e("ddms",
"Failed to reopen debug port for Selected ClIEnt to: " + mNewDebugSelectedPort);
Log.e("ddms", ioe);
mNewDebugSelectedPort = mDebugSelectedPort; // no retry
}

int count;
try {
count = mSelector.select();
} catch (IOException ioe) {
ioe.printStackTrace();
continue;
} catch (CancelledKeyException cke) {
continue;
}

if (count == 0) {
continue;
} //這裡代碼寫的不是很好，android開發網提示大家因為這個NIO是DDMS工作在PC端的還不明顯，這樣輪訓的在一個while中，效率不是很高，CPU很容易占用率很高。

Set<SelectionKey> keys = mSelector.selectedKeys();
Iterator<SelectionKey> iter = keys.iterator(); //使用迭代器獲取這個選擇鍵

while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove();

try {
if (key.attachment() instanceof Client) { //判斷收到的key的附件是否是ClIEnt的實例
processClIEntActivity(key);
}
else if (key.attachment() instanceof Debugger) { //如果是Debug實例
processDebuggerActivity(key);
}
else if (key.attachment() instanceof MonitorThread) {
processDebugSelectedActivity(key);
}
else {
Log.e("ddms", "unknown activity key");
}
} catch (Exception e) {
Log.e("ddms", "Exception during activity from Selector.");
Log.e("ddms", e);
}
}
} catch (Exception e) {
Log.e("ddms", "Exception MonitorThread.run()");
Log.e("ddms", e);
}
}
}

int getDebugSelectedPort() {
return mDebugSelectedPort;
}

private void processClIEntActivity(SelectionKey key) {
Client client = (ClIEnt)key.attachment();

try {
if (key.isReadable() == false || key.isValid() == false) {
Log.d("ddms", "Invalid key from " + client + ". Dropping clIEnt.");
dropClient(clIEnt, true /* notify */);
return;
}

clIEnt.read();

JdwpPacket packet = clIEnt.getJdwpPacket();
while (packet != null) {
if (packet.isDdmPacket()) {
// unsolicited DDM request - hand it off
assert !packet.isReply();
callHandler(clIEnt, packet, null);
packet.consume();
} else if (packet.isReply()
&& clIEnt.isResponseToUs(packet.getId()) != null) {
// reply to earlIEr DDM request
ChunkHandler handler = clIEnt
.isResponseToUs(packet.getId());
if (packet.isError())
clIEnt.packetFailed(packet);
else if (packet.isEmpty())
Log.d("ddms", "Got empty reply for 0x"
+ Integer.toHexString(packet.getId())
+ " from " + clIEnt);
else
callHandler(clIEnt, packet, handler);
packet.consume();
clIEnt.removeRequestId(packet.getId());
} else {
Log.v("ddms", "Forwarding clIEnt "
+ (packet.isReply() ? "reply" : "event") + " 0x"
+ Integer.toHexString(packet.getId()) + " to "
+ clIEnt.getDebugger());
clIEnt.forwardPacketToDebugger(packet);
}

packet = clIEnt.getJdwpPacket();
}
} catch (CancelledKeyException e) { //注意正確處理這個異常
dropClient(clIEnt, true /* notify */);
} catch (IOException ex) {
dropClient(clIEnt, true /* notify */);
} catch (Exception ex) {
Log.e("ddms", ex);

dropClient(clIEnt, true /* notify */);

if (ex instanceof BufferOverflowException) { //可能存在緩沖區異常
Log.w("ddms",
"ClIEnt data packet exceeded maximum buffer size "
+ clIEnt);
} else {
// don't know what this is, display it
Log.e("ddms", ex);
}
}
}

private void callHandler(Client clIEnt, JdwpPacket packet,
ChunkHandler handler) {

// on first DDM packet received, broadcast a "ready" message
if (!clIEnt.ddmSeen())
broadcast(CLIENT_READY, clIEnt);

ByteBuffer buf = packet.getPayload();
int type, length;
boolean reply = true;

type = buf.getInt();
length = buf.getInt();

if (handler == null) {
// not a reply, figure out who wants it
synchronized (mHandlerMap) {
handler = mHandlerMap.get(type);
reply = false;
}
}

if (handler == null) {
Log.w("ddms", "Received unsupported chunk type "
+ ChunkHandler.name(type) + " (len=" + length + ")");
} else {
Log.d("ddms", "Calling handler for " + ChunkHandler.name(type)
+ " [" + handler + "] (len=" + length + ")");
ByteBuffer ibuf = buf.slice();
ByteBuffer roBuf = ibuf.asReadOnlyBuffer(); // enforce R/O
roBuf.order(ChunkHandler.CHUNK_ORDER);
synchronized (mClIEntList) {
handler.handleChunk(clIEnt, type, roBuf, reply, packet.getId());
}
}
}

synchronized void dropClient(Client clIEnt, boolean notify) {
if (mInstance == null) {
return;
}

synchronized (mClIEntList) {
if (mClientList.remove(clIEnt) == false) {
return;
}
}
clIEnt.close(notify);
broadcast(CLIENT_DISCONNECTED, clIEnt);

/*
* http://forum.Java.sun.com/thread.JSPa?threadID=726715&start=0
* http://bugs.sun.com/bugdatabase/vIEw_bug.do?bug_id=5073504
*/
wakeup();
}

/*
* Process activity from one of the debugger sockets. This could be a new
* connection or a data packet.
*/
private void processDebuggerActivity(SelectionKey key) {
Debugger dbg = (Debugger)key.attachment();

try {
if (key.isAcceptable()) { //處理Server響應這個事件
try {
acceptNewDebugger(dbg, null);
} catch (IOException ioe) {
Log.w("ddms", "debugger accept() failed");
ioe.printStackTrace();
}
} else if (key.isReadable()) { //如果是收到的數據，則可讀取
processDebuggerData(key);
} else {
Log.d("ddm-debugger", "key in unknown state");
}
} catch (CancelledKeyException cke) { //記住，NIO處理這個異常，很多入門的開發者很容易忘記
// key has been cancelled we can ignore that.
}
}

private void acceptNewDebugger(Debugger dbg, ServerSocketChannel acceptChan) //這裡用到了阻塞方式
throws IOException {

synchronized (mClIEntList) {
SocketChannel chan;

if (acceptChan == null)
chan = dbg.accept();
else
chan = dbg.accept(acceptChan);

if (chan != null) {
chan.socket().setTcpNoDelay(true);

wakeup();

try {
chan.register(mSelector, SelectionKey.OP_READ, dbg);
} catch (IOException ioe) {
// failed, drop the connection
dbg.closeData();
throw ioe;
} catch (RuntimeException re) {
// failed, drop the connection
dbg.closeData();
throw re;
}
} else {
Log.w("ddms", "ignoring duplicate debugger");
}
}
}

private void processDebuggerData(SelectionKey key) {
Debugger dbg = (Debugger)key.attachment();

try {
dbg.read();

JdwpPacket packet = dbg.getJdwpPacket();
while (packet != null) {
Log.v("ddms", "Forwarding dbg req 0x"
+ Integer.toHexString(packet.getId()) + " to "
+ dbg.getClIEnt());

dbg.forwardPacketToClIEnt(packet);

packet = dbg.getJdwpPacket();
}
} catch (IOException ioe) {
Log.d("ddms", "Closing connection to debugger " + dbg);
dbg.closeData();
Client client = dbg.getClIEnt();
if (clIEnt.isDdmAware()) {
Log.d("ddms", " (recycling clIEnt connection as well)");

client.getDeviceImpl().getMonitor().addClientToDropAndReopen(clIEnt,
IDebugPortProvider.NO_STATIC_PORT);
} else {
Log.d("ddms", " (recycling clIEnt connection as well)");
// we should drop the clIEnt, but also attempt to reopen it.
// This is done by the DeviceMonitor.
client.getDeviceImpl().getMonitor().addClientToDropAndReopen(clIEnt,
IDebugPortProvider.NO_STATIC_PORT);
}
}
}

private void wakeup() {
mSelector.wakeup();
}

synchronized void quit() {
mQuit = true;
wakeup();
Log.d("ddms", "Waiting for Monitor thread");
try {
this.join();
// since we're quitting, lets drop all the clIEnt and disconnect
// the DebugSelectedPort
synchronized (mClIEntList) {
for (Client c : mClIEntList) {
c.close(false /* notify */);
broadcast(CLIENT_DISCONNECTED, c);
}
mClIEntList.clear();
}

if (mDebugSelectedChan != null) {
mDebugSelectedChan.close();
mDebugSelectedChan.socket().close();
mDebugSelectedChan = null;
}
mSelector.close();
} catch (InterruptedException IE) {
IE.printStackTrace();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

mInstance = null;
}

synchronized void addClient(Client clIEnt) {
if (mInstance == null) {
return;
}

Log.d("ddms", "Adding new client " + clIEnt);

synchronized (mClIEntList) {
mClientList.add(clIEnt);

try {
wakeup();

clIEnt.register(mSelector);

Debugger dbg = clIEnt.getDebugger();
if (dbg != null) {
dbg.registerListener(mSelector);
}
} catch (IOException ioe) {
// not really expecting this to happen
ioe.printStackTrace();
}
}
}

/*
* Broadcast an event to all message handlers.
*/
private void broadcast(int event, Client clIEnt) {
Log.d("ddms", "broadcast " + event + ": " + clIEnt);

/*
* The handler objects appear once in mHandlerMap for each message they
* handle. We want to notify them once each, so we convert the HashMap
* to a HashSet before we iterate.
*/
HashSet<ChunkHandler> set;
synchronized (mHandlerMap) {
Collection<ChunkHandler> values = mHandlerMap.values();
set = new HashSet<ChunkHandler>(values);
}

Iterator<ChunkHandler> iter = set.iterator();
while (iter.hasNext()) {
ChunkHandler handler = iter.next();
switch (event) {
case CLIENT_READY:
try {
handler.clientReady(clIEnt);
} catch (IOException ioe) {
// Something failed with the clIEnt. It should
// fall out of the list the next time we try to
// do something with it, so we discard the
// exception here and assume cleanup will happen
// later. May need to propagate farther. The
// trouble is that not all values for "event" may
// actually throw an exception.
Log.w("ddms",
"Got exception while broadcasting 'ready'");
return;
}
break;
case CLIENT_DISCONNECTED:
handler.clientDisconnected(clIEnt);
break;
default:
throw new UnsupportedOperationException();
}
}

}

/**
* Opens (or reopens) the "debug selected" port and listen for connections.
* @return true if the port was opened successfully.
* @throws IOException
*/
private boolean reopenDebugSelectedPort() throws IOException {

Log.d("ddms", "reopen debug-selected port: " + mNewDebugSelectedPort);
if (mDebugSelectedChan != null) {
mDebugSelectedChan.close();
}

mDebugSelectedChan = ServerSocketChannel.open();
mDebugSelectedChan.configureBlocking(false); // required for Selector

InetSocketAddress addr = new InetSocketAddress(
InetAddress.getByName("localhost"), //$NON-NLS-1$
mNewDebugSelectedPort);
mDebugSelectedChan.socket().setReuseAddress(true); // enable SO_REUSEADDR

try {
mDebugSelectedChan.socket().bind(addr);
if (mSelectedClIEnt != null) {
mSelectedClient.update(ClIEnt.CHANGE_PORT);
}

mDebugSelectedChan.register(mSelector, SelectionKey.OP_ACCEPT, this);

return true;
} catch (Java.Net.BindException e) {
displayDebugSelectedBindError(mNewDebugSelectedPort);

// do not attempt to reopen it.
mDebugSelectedChan = null;
mNewDebugSelectedPort = -1;

return false;
}
}

/*
* We have some activity on the "debug selected" port. Handle it.
*/
private void processDebugSelectedActivity(SelectionKey key) {
assert key.isAcceptable();

ServerSocketChannel acceptChan = (ServerSocketChannel)key.channel();

/*
* Find the debugger associated with the currently-selected clIEnt.
*/
if (mSelectedClIEnt != null) {
Debugger dbg = mSelectedClIEnt.getDebugger();

if (dbg != null) {
Log.d("ddms", "Accepting connection on 'debug selected' port");
try {
acceptNewDebugger(dbg, acceptChan);
} catch (IOException ioe) {
// clIEnt should be gone, keep going
}

return;
}
}

Log.w("ddms",
"Connection on 'debug selected' port, but none selected");
try {
SocketChannel chan = acceptChan.accept();
chan.close();
} catch (IOException ioe) {
// not expected; clIEnt should be gone, keep going
} catch (NotYetBoundException e) {
displayDebugSelectedBindError(mDebugSelectedPort);
}
}

private void displayDebugSelectedBindError(int port) {
String message = String.format(
"Could not open Selected VM debug port (%1$d). Make sure you do not have another instance of DDMS or of the eclipse plugin running. If it's being used by something else, choose a new port number in the preferences.",
port);

Log.logAndDisplay(LogLevel.ERROR, "ddms", message);
}
}

從上面來看android的開源代碼有關PC上的寫的不是很好，很多實現的地方都是用了嚴重的縫縫補補方式解決，有些習慣不是很到位，有關本NIO例子由於涉及的項目對象多，理解需要網友深入分析DDMS源碼中的每個對象。細節寫的不是很理想，android123推薦大家，畫出UML後再分析更清晰。