Android6.0 消息机制原理解析

 mHandlerThread = new ServiceThread(TAG,        Process.THREAD_PRIORITY_DISPLAY, false /*allowIo*/);    mHandlerThread.start();    mHandler = new PowerManagerHandler(mHandlerThread.getLooper()); 

   private final class PowerManagerHandler extends Handler {    public PowerManagerHandler(Looper looper) {      super(looper, null, true /*async*/);    }    @Override    public void handleMessage(Message msg) {      switch (msg.what) {        case MSG_USER_ACTIVITY_TIMEOUT:          handleUserActivityTimeout();          break;        case MSG_SANDMAN:          handleSandman();          break;        case MSG_SCREEN_BRIGHTNESS_BOOST_TIMEOUT:          handleScreenBrightnessBoostTimeout();          break;        case MSG_CHECK_WAKE_LOCK_ACQUIRE_TIMEOUT:          checkWakeLockAquireTooLong();          Message m = mHandler.obtainMessage(MSG_CHECK_WAKE_LOCK_ACQUIRE_TIMEOUT);          m.setAsynchronous(true);          mHandler.sendMessageDelayed(m, WAKE_LOCK_ACQUIRE_TOO_LONG_TIMEOUT);          break;      }    }  }

 public void run() {    mTid = Process.myTid();    Looper.prepare();    synchronized (this) {      mLooper = Looper.myLooper();//赋值后notifyall，主要是getLooper函数返回的是mLooper      notifyAll();    }    Process.setThreadPriority(mPriority);    onLooperPrepared();    Looper.loop();    mTid = -1;  }

public static void prepare() {    prepare(true);  }  private static void prepare(boolean quitAllowed) {    if (sThreadLocal.get() != null) {      throw new RuntimeException("Only one Looper may be created per thread");    }    sThreadLocal.set(new Looper(quitAllowed));  } 

   private Looper(boolean quitAllowed) {    mQueue = new MessageQueue(quitAllowed);    mThread = Thread.currentThread();  } 

MessageQueue(boolean quitAllowed) {    mQuitAllowed = quitAllowed;    mPtr = nativeInit();  } 

 static jlong android_os_MessageQueue_nativeInit(JNIEnv* env, jclass clazz) {  NativeMessageQueue* nativeMessageQueue = new NativeMessageQueue();  if (!nativeMessageQueue) {    jniThrowRuntimeException(env, "Unable to allocate native queue");    return 0;  }  nativeMessageQueue->incStrong(env);  return reinterpret_cast<jlong>(nativeMessageQueue);} 

NativeMessageQueue::NativeMessageQueue() :    mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {  mLooper = Looper::getForThread();  if (mLooper == NULL) {    mLooper = new Looper(false);    Looper::setForThread(mLooper);  }}

 Looper::Looper(bool allowNonCallbacks) :    mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),    mPolling(false), mEpollFd(-1), mEpollRebuildRequired(false),    mNextRequestSeq(0), mResponseIndex(0), mNextMessageUptime(LLONG_MAX) {  mWakeEventFd = eventfd(0, EFD_NONBLOCK);  LOG_ALWAYS_FATAL_IF(mWakeEventFd < 0, "Could not make wake event fd. errno=%d", errno);  AutoMutex _l(mLock);  rebuildEpollLocked();}

 void Looper::rebuildEpollLocked() {  // Close old epoll instance if we have one.  if (mEpollFd >= 0) {#if DEBUG_CALLBACKS    ALOGD("%p ~ rebuildEpollLocked - rebuilding epoll set", this);#endif    close(mEpollFd);  }  // Allocate the new epoll instance and register the wake pipe.  mEpollFd = epoll_create(EPOLL_SIZE_HINT);  LOG_ALWAYS_FATAL_IF(mEpollFd < 0, "Could not create epoll instance. errno=%d", errno);  struct epoll_event eventItem;  memset(& eventItem, 0, sizeof(epoll_event)); // zero out unused members of data field union  eventItem.events = EPOLLIN;  eventItem.data.fd = mWakeEventFd;  int result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeEventFd, & eventItem);  LOG_ALWAYS_FATAL_IF(result != 0, "Could not add wake event fd to epoll instance. errno=%d",      errno);  for (size_t i = 0; i < mRequests.size(); i++) {    const Request& request = mRequests.valueAt(i);    struct epoll_event eventItem;    request.initEventItem(&eventItem);    int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, request.fd, & eventItem);    if (epollResult < 0) {      ALOGE("Error adding epoll events for fd %d while rebuilding epoll set, errno=%d",          request.fd, errno);    }  }} 

public void run() {    mTid = Process.myTid();    Looper.prepare();    synchronized (this) {      mLooper = Looper.myLooper();      notifyAll();    }    Process.setThreadPriority(mPriority);    onLooperPrepared();    Looper.loop();    mTid = -1;  } 

public static void loop() {    final Looper me = myLooper();    if (me == null) {      throw new RuntimeException("No Looper; Looper.prepare() wasn't called on this thread.");    }    final MessageQueue queue = me.mQueue;//得到Looper的mQueue    // Make sure the identity of this thread is that of the local process,    // and keep track of what that identity token actually is.    Binder.clearCallingIdentity();    final long ident = Binder.clearCallingIdentity();    for (;;) {      Message msg = queue.next(); // might block这个函数会阻塞，阻塞主要是epoll_wait      if (msg == null) {        // No message indicates that the message queue is quitting.        return;      }      // This must be in a local variable, in case a UI event sets the logger      Printer logging = me.mLogging;//自己打的打印      if (logging != null) {        logging.println(">>>>> Dispatching to " + msg.target + " " +            msg.callback + ": " + msg.what);      }      msg.target.dispatchMessage(msg);      if (logging != null) {        logging.println("<<<<< Finished to " + msg.target + " " + msg.callback);      }      // Make sure that during the course of dispatching the      // identity of the thread wasn't corrupted.      final long newIdent = Binder.clearCallingIdentity();      if (ident != newIdent) {        Log.wtf(TAG, "Thread identity changed from 0x"            + Long.toHexString(ident) + " to 0x"            + Long.toHexString(newIdent) + " while dispatching to "            + msg.target.getClass().getName() + " "            + msg.callback + " what=" + msg.what);      }      msg.recycleUnchecked();    }  }

Message next() {    // Return here if the message loop has already quit and been disposed.    // This can happen if the application tries to restart a looper after quit    // which is not supported.    final long ptr = mPtr;//之前保留的指针    if (ptr == 0) {      return null;    }    int pendingIdleHandlerCount = -1; // -1 only during first iteration    int nextPollTimeoutMillis = 0;    for (;;) {      if (nextPollTimeoutMillis != 0) {        Binder.flushPendingCommands();      }      nativePollOnce(ptr, nextPollTimeoutMillis); 

static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj,    jlong ptr, jint timeoutMillis) {  NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);  nativeMessageQueue->pollOnce(env, obj, timeoutMillis);}

 int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) {  int result = 0;  for (;;) {    while (mResponseIndex < mResponses.size()) {      const Response& response = mResponses.itemAt(mResponseIndex++);      int ident = response.request.ident;      if (ident >= 0) {        int fd = response.request.fd;        int events = response.events;        void* data = response.request.data;#if DEBUG_POLL_AND_WAKE        ALOGD("%p ~ pollOnce - returning signalled identifier %d: "            "fd=%d, events=0x%x, data=%p",            this, ident, fd, events, data);#endif        if (outFd != NULL) *outFd = fd;        if (outEvents != NULL) *outEvents = events;        if (outData != NULL) *outData = data;        return ident;      }    }    if (result != 0) {#if DEBUG_POLL_AND_WAKE      ALOGD("%p ~ pollOnce - returning result %d", this, result);#endif      if (outFd != NULL) *outFd = 0;      if (outEvents != NULL) *outEvents = 0;      if (outData != NULL) *outData = NULL;      return result;    }    result = pollInner(timeoutMillis);  }} 

int Looper::pollInner(int timeoutMillis) {#if DEBUG_POLL_AND_WAKE  ALOGD("%p ~ pollOnce - waiting: timeoutMillis=%d", this, timeoutMillis);#endif  // Adjust the timeout based on when the next message is due.  if (timeoutMillis != 0 && mNextMessageUptime != LLONG_MAX) {    nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);    int messageTimeoutMillis = toMillisecondTimeoutDelay(now, mNextMessageUptime);    if (messageTimeoutMillis >= 0        && (timeoutMillis < 0 || messageTimeoutMillis < timeoutMillis)) {      timeoutMillis = messageTimeoutMillis;    }#if DEBUG_POLL_AND_WAKE    ALOGD("%p ~ pollOnce - next message in %" PRId64 "ns, adjusted timeout: timeoutMillis=%d",        this, mNextMessageUptime - now, timeoutMillis);#endif  }  // Poll.  int result = POLL_WAKE;  mResponses.clear();//清空mResponses  mResponseIndex = 0;  // We are about to idle.  mPolling = true;  struct epoll_event eventItems[EPOLL_MAX_EVENTS];  int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis);//epoll_wait主要线程阻塞在这，这个阻塞的时间也是有java层传过来的  // No longer idling.  mPolling = false;  // Acquire lock.  mLock.lock();  // Rebuild epoll set if needed.  if (mEpollRebuildRequired) {    mEpollRebuildRequired = false;    rebuildEpollLocked();    goto Done;  }  // Check for poll error.  if (eventCount < 0) {    if (errno == EINTR) {      goto Done;    }    ALOGW("Poll failed with an unexpected error, errno=%d", errno);    result = POLL_ERROR;    goto Done;  }  // Check for poll timeout.  if (eventCount == 0) {#if DEBUG_POLL_AND_WAKE    ALOGD("%p ~ pollOnce - timeout", this);#endif    result = POLL_TIMEOUT;    goto Done;  }  // Handle all events.#if DEBUG_POLL_AND_WAKE  ALOGD("%p ~ pollOnce - handling events from %d fds", this, eventCount);#endif  for (int i = 0; i < eventCount; i++) {    int fd = eventItems[i].data.fd;    uint32_t epollEvents = eventItems[i].events;    if (fd == mWakeEventFd) {//通知唤醒线程的事件      if (epollEvents & EPOLLIN) {        awoken();      } else {        ALOGW("Ignoring unexpected epoll events 0x%x on wake event fd.", epollEvents);      }    } else {      ssize_t requestIndex = mRequests.indexOfKey(fd);//之前addFd的事件      if (requestIndex >= 0) {        int events = 0;        if (epollEvents & EPOLLIN) events |= EVENT_INPUT;        if (epollEvents & EPOLLOUT) events |= EVENT_OUTPUT;        if (epollEvents & EPOLLERR) events |= EVENT_ERROR;        if (epollEvents & EPOLLHUP) events |= EVENT_HANGUP;        pushResponse(events, mRequests.valueAt(requestIndex));//放在mResponses中      } else {        ALOGW("Ignoring unexpected epoll events 0x%x on fd %d that is "            "no longer registered.", epollEvents, fd);      }    }  }Done: ;  // Invoke pending message callbacks.  mNextMessageUptime = LLONG_MAX;  while (mMessageEnvelopes.size() != 0) {// 这块主要是c层的消息，java层的消息是自己管理的    nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);    const MessageEnvelope& messageEnvelope = mMessageEnvelopes.itemAt(0);    if (messageEnvelope.uptime <= now) {      // Remove the envelope from the list.      // We keep a strong reference to the handler until the call to handleMessage      // finishes. Then we drop it so that the handler can be deleted *before*      // we reacquire our lock.      { // obtain handler        sp<MessageHandler> handler = messageEnvelope.handler;        Message message = messageEnvelope.message;        mMessageEnvelopes.removeAt(0);        mSendingMessage = true;        mLock.unlock();#if DEBUG_POLL_AND_WAKE || DEBUG_CALLBACKS        ALOGD("%p ~ pollOnce - sending message: handler=%p, what=%d",            this, handler.get(), message.what);#endif        handler->handleMessage(message);      } // release handler      mLock.lock();      mSendingMessage = false;      result = POLL_CALLBACK;    } else {      // The last message left at the head of the queue determines the next wakeup time.      mNextMessageUptime = messageEnvelope.uptime;      break;    }  }  // Release lock.  mLock.unlock();  // Invoke all response callbacks.  for (size_t i = 0; i < mResponses.size(); i++) {//这是之前addFd的事件的处理，主要是遍历mResponses，然后调用其回调    Response& response = mResponses.editItemAt(i);    if (response.request.ident == POLL_CALLBACK) {      int fd = response.request.fd;      int events = response.events;      void* data = response.request.data;#if DEBUG_POLL_AND_WAKE || DEBUG_CALLBACKS      ALOGD("%p ~ pollOnce - invoking fd event callback %p: fd=%d, events=0x%x, data=%p",          this, response.request.callback.get(), fd, events, data);#endif      // Invoke the callback. Note that the file descriptor may be closed by      // the callback (and potentially even reused) before the function returns so      // we need to be a little careful when removing the file descriptor afterwards.      int callbackResult = response.request.callback->handleEvent(fd, events, data);      if (callbackResult == 0) {        removeFd(fd, response.request.seq);      }      // Clear the callback reference in the response structure promptly because we      // will not clear the response vector itself until the next poll.      response.request.callback.clear();      result = POLL_CALLBACK;    }  }  return result;} 

     for (;;) {      Message msg = queue.next(); // might block      if (msg == null) {        // No message indicates that the message queue is quitting.        return;      }      // This must be in a local variable, in case a UI event sets the logger      Printer logging = me.mLogging;//自己的打印      if (logging != null) {        logging.println(">>>>> Dispatching to " + msg.target + " " +            msg.callback + ": " + msg.what);      }      msg.target.dispatchMessage(msg);      if (logging != null) {        logging.println("<<<<< Finished to " + msg.target + " " + msg.callback);      }      // Make sure that during the course of dispatching the      // identity of the thread wasn't corrupted.      final long newIdent = Binder.clearCallingIdentity();      if (ident != newIdent) {        Log.wtf(TAG, "Thread identity changed from 0x"            + Long.toHexString(ident) + " to 0x"            + Long.toHexString(newIdent) + " while dispatching to "            + msg.target.getClass().getName() + " "            + msg.callback + " what=" + msg.what);      }      msg.recycleUnchecked();    }  }

public void setMessageLogging(@Nullable Printer printer) {    mLogging = printer;  } Printer就是一个interface public interface Printer {  /**   * Write a line of text to the output. There is no need to terminate   * the given string with a newline.   */  void println(String x);}

 Message msg = mHandler.obtainMessage(MSG_CHECK_WAKE_LOCK_ACQUIRE_TIMEOUT); msg.setAsynchronous(true); mHandler.sendMessageDelayed(msg, WAKE_LOCK_ACQUIRE_TOO_LONG_TIMEOUT); 

public final Message obtainMessage(int what)  {    return Message.obtain(this, what);  } 

public static Message obtain(Handler h, int what) {    Message m = obtain();//就是新建一个Message    m.target = h;    m.what = what;    return m;  } 

   public void dispatchMessage(Message msg) {    if (msg.callback != null) {      handleCallback(msg);//这种就是用post形式发送，带Runnable的    } else {      if (mCallback != null) {//这种是handler传参的时候就是传入了mCallback回调了        if (mCallback.handleMessage(msg)) {          return;        }      }      handleMessage(msg);//最后就是在自己实现的handleMessage处理    }  }

   public boolean sendMessageAtTime(Message msg, long uptimeMillis) {    MessageQueue queue = mQueue;    if (queue == null) {      RuntimeException e = new RuntimeException(          this + " sendMessageAtTime() called with no mQueue");      Log.w("Looper", e.getMessage(), e);      return false;    }    return enqueueMessage(queue, msg, uptimeMillis);  } 

private boolean enqueueMessage(MessageQueue queue, Message msg, long uptimeMillis) {    msg.target = this;    if (mAsynchronous) {      msg.setAsynchronous(true);    }    return queue.enqueueMessage(msg, uptimeMillis);  } 

boolean enqueueMessage(Message msg, long when) {    if (msg.target == null) {      throw new IllegalArgumentException("Message must have a target.");    }    if (msg.isInUse()) {      throw new IllegalStateException(msg + " This message is already in use.");    }    synchronized (this) {      if (mQuitting) {        IllegalStateException e = new IllegalStateException(            msg.target + " sending message to a Handler on a dead thread");        Log.w(TAG, e.getMessage(), e);        msg.recycle();        return false;      }      msg.markInUse();      msg.when = when;      Message p = mMessages;      boolean needWake;      if (p == null || when == 0 || when < p.when) {        // New head, wake up the event queue if blocked.        msg.next = p;        mMessages = msg;        needWake = mBlocked;      } else {        // Inserted within the middle of the queue. Usually we don't have to wake        // up the event queue unless there is a barrier at the head of the queue        // and the message is the earliest asynchronous message in the queue.        needWake = mBlocked && p.target == null && msg.isAsynchronous();        Message prev;        for (;;) {          prev = p;          p = p.next;          if (p == null || when < p.when) {            break;          }          if (needWake && p.isAsynchronous()) {            needWake = false;          }        }        msg.next = p; // invariant: p == prev.next        prev.next = msg;      }      // We can assume mPtr != 0 because mQuitting is false.      if (needWake) {        nativeWake(mPtr);      }    }    return true;  } 

 static void android_os_MessageQueue_nativeWake(JNIEnv* env, jclass clazz, jlong ptr) {  NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);  nativeMessageQueue->wake();}void NativeMessageQueue::wake() {  mLooper->wake();} 

void Looper::wake() {#if DEBUG_POLL_AND_WAKE  ALOGD("%p ~ wake", this);#endif  uint64_t inc = 1;  ssize_t nWrite = TEMP_FAILURE_RETRY(write(mWakeEventFd, &inc, sizeof(uint64_t)));  if (nWrite != sizeof(uint64_t)) {    if (errno != EAGAIN) {      ALOGW("Could not write wake signal, errno=%d", errno);    }  }}

 void Looper::sendMessageAtTime(nsecs_t uptime, const sp<MessageHandler>& handler,    const Message& message) {#if DEBUG_CALLBACKS  ALOGD("%p ~ sendMessageAtTime - uptime=%" PRId64 ", handler=%p, what=%d",      this, uptime, handler.get(), message.what);#endif  size_t i = 0;  { // acquire lock    AutoMutex _l(mLock);    size_t messageCount = mMessageEnvelopes.size();    while (i < messageCount && uptime >= mMessageEnvelopes.itemAt(i).uptime) {      i += 1;    }    MessageEnvelope messageEnvelope(uptime, handler, message);    mMessageEnvelopes.insertAt(messageEnvelope, i, 1);    // Optimization: If the Looper is currently sending a message, then we can skip    // the call to wake() because the next thing the Looper will do after processing    // messages is to decide when the next wakeup time should be. In fact, it does    // not even matter whether this code is running on the Looper thread.    if (mSendingMessage) {      return;    }  } // release lock  // Wake the poll loop only when we enqueue a new message at the head.  if (i == 0) {    wake();  }} 

   while (mMessageEnvelopes.size() != 0) {    nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);    const MessageEnvelope& messageEnvelope = mMessageEnvelopes.itemAt(0);    if (messageEnvelope.uptime <= now) {      // Remove the envelope from the list.      // We keep a strong reference to the handler until the call to handleMessage      // finishes. Then we drop it so that the handler can be deleted *before*      // we reacquire our lock.      { // obtain handler        sp<MessageHandler> handler = messageEnvelope.handler;        Message message = messageEnvelope.message;        mMessageEnvelopes.removeAt(0);        mSendingMessage = true;        mLock.unlock();#if DEBUG_POLL_AND_WAKE || DEBUG_CALLBACKS        ALOGD("%p ~ pollOnce - sending message: handler=%p, what=%d",            this, handler.get(), message.what);#endif        handler->handleMessage(message);      } // release handler      mLock.lock();      mSendingMessage = false;      result = POLL_CALLBACK;    } else {      // The last message left at the head of the queue determines the next wakeup time.      mNextMessageUptime = messageEnvelope.uptime;      break;    }  } 

   sp<StubMessageHandler> handler = new StubMessageHandler();  mLooper->sendMessageAtTime(now + ms2ns(100), handler, Message(MSG_TEST1)); StubMessageHandler继承MessageHandler就必须实现handleMessage方法 class StubMessageHandler : public MessageHandler {public:  Vector<Message> messages;  virtual void handleMessage(const Message& message) {    messages.push(message);  }}; 

 struct Message {  Message() : what(0) { }  Message(int what) : what(what) { }  /* The message type. (interpretation is left up to the handler) */  int what;};/** * Interface for a Looper message handler. * * The Looper holds a strong reference to the message handler whenever it has * a message to deliver to it. Make sure to call Looper::removeMessages * to remove any pending messages destined for the handler so that the handler * can be destroyed. */class MessageHandler : public virtual RefBase {protected:  virtual ~MessageHandler() { }public:  /**   * Handles a message.   */  virtual void handleMessage(const Message& message) = 0;};

 int Looper::addFd(int fd, int ident, int events, Looper_callbackFunc callback, void* data) {  return addFd(fd, ident, events, callback ? new SimpleLooperCallback(callback) : NULL, data);}int Looper::addFd(int fd, int ident, int events, const sp<LooperCallback>& callback, void* data) {#if DEBUG_CALLBACKS  ALOGD("%p ~ addFd - fd=%d, ident=%d, events=0x%x, callback=%p, data=%p", this, fd, ident,      events, callback.get(), data);#endif  if (!callback.get()) {    if (! mAllowNonCallbacks) {      ALOGE("Invalid attempt to set NULL callback but not allowed for this looper.");      return -1;    }    if (ident < 0) {      ALOGE("Invalid attempt to set NULL callback with ident < 0.");      return -1;    }  } else {    ident = POLL_CALLBACK;  }  { // acquire lock    AutoMutex _l(mLock);    Request request;    request.fd = fd;    request.ident = ident;    request.events = events;    request.seq = mNextRequestSeq++;    request.callback = callback;    request.data = data;    if (mNextRequestSeq == -1) mNextRequestSeq = 0; // reserve sequence number -1    struct epoll_event eventItem;    request.initEventItem(&eventItem);    ssize_t requestIndex = mRequests.indexOfKey(fd);    if (requestIndex < 0) {      int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, fd, & eventItem);//加入epoll      if (epollResult < 0) {        ALOGE("Error adding epoll events for fd %d, errno=%d", fd, errno);        return -1;      }      mRequests.add(fd, request);//放入mRequests中    } else {      int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_MOD, fd, & eventItem);//更新      if (epollResult < 0) {        if (errno == ENOENT) {          // Tolerate ENOENT because it means that an older file descriptor was          // closed before its callback was unregistered and meanwhile a new          // file descriptor with the same number has been created and is now          // being registered for the first time. This error may occur naturally          // when a callback has the side-effect of closing the file descriptor          // before returning and unregistering itself. Callback sequence number          // checks further ensure that the race is benign.          //          // Unfortunately due to kernel limitations we need to rebuild the epoll          // set from scratch because it may contain an old file handle that we are          // now unable to remove since its file descriptor is no longer valid.          // No such problem would have occurred if we were using the poll system          // call instead, but that approach carries others disadvantages.#if DEBUG_CALLBACKS          ALOGD("%p ~ addFd - EPOLL_CTL_MOD failed due to file descriptor "              "being recycled, falling back on EPOLL_CTL_ADD, errno=%d",              this, errno);#endif          epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, fd, & eventItem);          if (epollResult < 0) {            ALOGE("Error modifying or adding epoll events for fd %d, errno=%d",                fd, errno);            return -1;          }          scheduleEpollRebuildLocked();        } else {          ALOGE("Error modifying epoll events for fd %d, errno=%d", fd, errno);          return -1;        }      }      mRequests.replaceValueAt(requestIndex, request);    }  } // release lock  return 1;} 

     } else {      ssize_t requestIndex = mRequests.indexOfKey(fd);      if (requestIndex >= 0) {        int events = 0;        if (epollEvents & EPOLLIN) events |= EVENT_INPUT;        if (epollEvents & EPOLLOUT) events |= EVENT_OUTPUT;        if (epollEvents & EPOLLERR) events |= EVENT_ERROR;        if (epollEvents & EPOLLHUP) events |= EVENT_HANGUP;        pushResponse(events, mRequests.valueAt(requestIndex));      } else {        ALOGW("Ignoring unexpected epoll events 0x%x on fd %d that is "            "no longer registered.", epollEvents, fd);      }    } 

   for (size_t i = 0; i < mResponses.size(); i++) {    Response& response = mResponses.editItemAt(i);    if (response.request.ident == POLL_CALLBACK) {      int fd = response.request.fd;      int events = response.events;      void* data = response.request.data;#if DEBUG_POLL_AND_WAKE || DEBUG_CALLBACKS      ALOGD("%p ~ pollOnce - invoking fd event callback %p: fd=%d, events=0x%x, data=%p",          this, response.request.callback.get(), fd, events, data);#endif      // Invoke the callback. Note that the file descriptor may be closed by      // the callback (and potentially even reused) before the function returns so      // we need to be a little careful when removing the file descriptor afterwards.      int callbackResult = response.request.callback->handleEvent(fd, events, data);      if (callbackResult == 0) {        removeFd(fd, response.request.seq);      }      // Clear the callback reference in the response structure promptly because we      // will not clear the response vector itself until the next poll.      response.request.callback.clear();      result = POLL_CALLBACK;    }  } 

   Pipe pipe;  StubCallbackHandler handler(true);  handler.setCallback(mLooper, pipe.receiveFd, Looper::EVENT_INPUT); 

class CallbackHandler {public:  void setCallback(const sp<Looper>& looper, int fd, int events) {    looper->addFd(fd, 0, events, staticHandler, this);//就是调用了looper的addFd函数，并且回调  }protected:  virtual ~CallbackHandler() { }  virtual int handler(int fd, int events) = 0;private:  static int staticHandler(int fd, int events, void* data) {//这个就是回调函数    return static_cast<CallbackHandler*>(data)->handler(fd, events);  }};class StubCallbackHandler : public CallbackHandler {public:  int nextResult;  int callbackCount;  int fd;  int events;  StubCallbackHandler(int nextResult) : nextResult(nextResult),      callbackCount(0), fd(-1), events(-1) {  }protected:  virtual int handler(int fd, int events) {//这个是通过回调函数再调到这里的    callbackCount += 1;    this->fd = fd;    this->events = events;    return nextResult;  }}; 

 int Looper::addFd(int fd, int ident, int events, Looper_callbackFunc callback, void* data) {  return addFd(fd, ident, events, callback ? new SimpleLooperCallback(callback) : NULL, data);} 

 class SimpleLooperCallback : public LooperCallback {protected:  virtual ~SimpleLooperCallback();public:  SimpleLooperCallback(Looper_callbackFunc callback);  virtual int handleEvent(int fd, int events, void* data);private:  Looper_callbackFunc mCallback;};SimpleLooperCallback::SimpleLooperCallback(Looper_callbackFunc callback) :    mCallback(callback) {}SimpleLooperCallback::~SimpleLooperCallback() {}int SimpleLooperCallback::handleEvent(int fd, int events, void* data) {  return mCallback(fd, events, data);} 

   public void addOnFileDescriptorEventListener(@NonNull FileDescriptor fd,      @OnFileDescriptorEventListener.Events int events,      @NonNull OnFileDescriptorEventListener listener) {    if (fd == null) {      throw new IllegalArgumentException("fd must not be null");    }    if (listener == null) {      throw new IllegalArgumentException("listener must not be null");    }    synchronized (this) {      updateOnFileDescriptorEventListenerLocked(fd, events, listener);    }  }

   public interface OnFileDescriptorEventListener {    public static final int EVENT_INPUT = 1 << 0;    public static final int EVENT_OUTPUT = 1 << 1;    public static final int EVENT_ERROR = 1 << 2;    /** @hide */    @Retention(RetentionPolicy.SOURCE)    @IntDef(flag=true, value={EVENT_INPUT, EVENT_OUTPUT, EVENT_ERROR})    public @interface Events {}    @Events int onFileDescriptorEvents(@NonNull FileDescriptor fd, @Events int events);  }

 private void updateOnFileDescriptorEventListenerLocked(FileDescriptor fd, int events,      OnFileDescriptorEventListener listener) {    final int fdNum = fd.getInt$();    int index = -1;    FileDescriptorRecord record = null;    if (mFileDescriptorRecords != null) {      index = mFileDescriptorRecords.indexOfKey(fdNum);      if (index >= 0) {        record = mFileDescriptorRecords.valueAt(index);        if (record != null && record.mEvents == events) {          return;        }      }    }    if (events != 0) {      events |= OnFileDescriptorEventListener.EVENT_ERROR;      if (record == null) {        if (mFileDescriptorRecords == null) {          mFileDescriptorRecords = new SparseArray<FileDescriptorRecord>();        }        record = new FileDescriptorRecord(fd, events, listener);//fd保存在FileDescriptorRecord对象        mFileDescriptorRecords.put(fdNum, record);//mFileDescriptorRecords然后保存在      } else {        record.mListener = listener;        record.mEvents = events;        record.mSeq += 1;      }      nativeSetFileDescriptorEvents(mPtr, fdNum, events);//调用native函数    } else if (record != null) {      record.mEvents = 0;      mFileDescriptorRecords.removeAt(index);    }  } 

static void android_os_MessageQueue_nativeSetFileDescriptorEvents(JNIEnv* env, jclass clazz,    jlong ptr, jint fd, jint events) {  NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);  nativeMessageQueue->setFileDescriptorEvents(fd, events);}

 void NativeMessageQueue::setFileDescriptorEvents(int fd, int events) {  if (events) {    int looperEvents = 0;    if (events & CALLBACK_EVENT_INPUT) {      looperEvents |= Looper::EVENT_INPUT;    }    if (events & CALLBACK_EVENT_OUTPUT) {      looperEvents |= Looper::EVENT_OUTPUT;    }    mLooper->addFd(fd, Looper::POLL_CALLBACK, looperEvents, this,        reinterpret_cast<void*>(events));  } else {    mLooper->removeFd(fd);  }}

 class NativeMessageQueue : public MessageQueue, public LooperCallback {public:  NativeMessageQueue();  virtual ~NativeMessageQueue();  virtual void raiseException(JNIEnv* env, const char* msg, jthrowable exceptionObj);  void pollOnce(JNIEnv* env, jobject obj, int timeoutMillis);  void wake();  void setFileDescriptorEvents(int fd, int events);  virtual int handleEvent(int fd, int events, void* data);

 int NativeMessageQueue::handleEvent(int fd, int looperEvents, void* data) {  int events = 0;  if (looperEvents & Looper::EVENT_INPUT) {    events |= CALLBACK_EVENT_INPUT;  }  if (looperEvents & Looper::EVENT_OUTPUT) {    events |= CALLBACK_EVENT_OUTPUT;  }  if (looperEvents & (Looper::EVENT_ERROR | Looper::EVENT_HANGUP | Looper::EVENT_INVALID)) {    events |= CALLBACK_EVENT_ERROR;  }  int oldWatchedEvents = reinterpret_cast<intptr_t>(data);  int newWatchedEvents = mPollEnv->CallIntMethod(mPollObj,      gMessageQueueClassInfo.dispatchEvents, fd, events); //调用回调  if (!newWatchedEvents) {    return 0; // unregister the fd  }  if (newWatchedEvents != oldWatchedEvents) {    setFileDescriptorEvents(fd, newWatchedEvents);  }  return 1;}

// Called from native code.  private int dispatchEvents(int fd, int events) {    // Get the file descriptor record and any state that might change.    final FileDescriptorRecord record;    final int oldWatchedEvents;    final OnFileDescriptorEventListener listener;    final int seq;    synchronized (this) {      record = mFileDescriptorRecords.get(fd);//通过fd得到FileDescriptorRecord       if (record == null) {        return 0; // spurious, no listener registered      }      oldWatchedEvents = record.mEvents;      events &= oldWatchedEvents; // filter events based on current watched set      if (events == 0) {        return oldWatchedEvents; // spurious, watched events changed      }      listener = record.mListener;      seq = record.mSeq;    }    // Invoke the listener outside of the lock.    int newWatchedEvents = listener.onFileDescriptorEvents(//listener回调        record.mDescriptor, events);    if (newWatchedEvents != 0) {      newWatchedEvents |= OnFileDescriptorEventListener.EVENT_ERROR;    }    // Update the file descriptor record if the listener changed the set of    // events to watch and the listener itself hasn't been updated since.    if (newWatchedEvents != oldWatchedEvents) {      synchronized (this) {        int index = mFileDescriptorRecords.indexOfKey(fd);        if (index >= 0 && mFileDescriptorRecords.valueAt(index) == record            && record.mSeq == seq) {          record.mEvents = newWatchedEvents;          if (newWatchedEvents == 0) {            mFileDescriptorRecords.removeAt(index);          }        }      }    }    // Return the new set of events to watch for native code to take care of.    return newWatchedEvents;  }