4.5MessageQueue解析

目录介绍

• 01.MessageQueue作用介绍
  • 1.1 主要是做什么
  • 1.2 数据结构是什么
• 02.MessageQueue的定义
  • 2.1 定义是什么
  • 2.2 从哪里创建的
• 03.enqueueMessage()源码
  • 3.1 解读源码流程
  • 3.2 如何处理消息延时
• 04.next()方法源码分析
  • 4.1 源码流程分析
  • 4.2 处理延迟消息逻辑
• 05.延迟发送消息队列分析

01.MessageQueue作用介绍

1.1 主要是做什么

• MessageQueue，主要包含2个操作：插入和读取。
  • 读取操作会伴随着删除操作，插入和读取对应的方法分别为enqueueMessage和next，其中enqueueMessage的作用是往消息队列中插入一条消息，而next的作用是从消息队列中取出一条消息并将其从消息队列中移除。

1.2 数据结构是什么

• 虽然MessageQueue叫消息队列，但是它的内部实现并不是用的队列。
  • 实际上它是通过一个单链表的数据结构来维护消息列表，单链表在插入和删除上比较有优势。

02.MessageQueue的定义

2.1 定义是什么

• 通过源码我们可以知道，MessageQueue维护了一个消息列表。
  • Message并不是直接添加到MessageQueue中，而是通过和Looper相关联的Handler来添加的。
  • 在当前线程中可以通过调用Looper.myQueue()方法来获取当前线程的MessageQueue。博客

  /**
   * Low-level class holding the list of messages to be dispatched by a
   * {@link Looper}.  Messages are not added directly to a MessageQueue,
   * but rather through {@link Handler} objects associated with the Looper.
   * 
   * <p>You can retrieve the MessageQueue for the current thread with
   * {@link Looper#myQueue() Looper.myQueue()}.
   */
  public final class MessageQueue

2.2 从哪里创建的

• 既然 Looper 对象已经由系统来为我们初始化好了，那我们就可以从中得到 mQueue对象。

  public Handler(Callback callback, boolean async) {
      //获取 MessageQueue 对象
      mQueue = mLooper.mQueue;
  }

• mQueue 又是在 Looper 类的构造函数中初始化的，且 mQueue 是 Looper 类的成员常量，这说明 Looper 与 MessageQueue 是一一对应的关系

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

03.enqueueMessage()源码

3.1 解读源码流程

• 源码如下所示
  • 在Message的源码中定义了一个成员属性target，其类型为Handler。由上面enqueuMessage的源码，我们可以看到，当Message没有处理其的Handler或该Message正在被处理的时候，都不能正常进入MessageQueue，这一点也是很容易理解的。
  • 当线程处于死亡状态的时候，Message会被回收掉，而不再进入该线程对应的MessageQueue中。否则，一切正常，enqueMessage就执行单链表的插入操作，将Message插入到MessageQueue中。

  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.
              //p为null(代表MessageQueue没有消息） 或者msg的触发时间是队列中最早的， 则进入该该分支
              //when什么时候会等于0，在调用sendMessageAtFrontOfQueue时      
              msg.next = p;
              mMessages = msg;
              needWake = mBlocked;
          } else {
              //将消息按时间顺序插入到MessageQueue。一般地，不需要唤醒事件队列，除非
              //消息队头存在阻塞，并且同时Message是队列中最早的异步消息
              // 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;
  }

3.2 如何处理消息延时

• 思考如何处理消息延时
  • MessageQueue是按照Message触发时间的先后顺序排列的【也就是说会改变链表的顺序】，队头的消息是将要最早触发的消息。
  • 排在越前面的越早触发，那我们现在应该了解到了，这个所谓的延时呢，不是延时发送消息，而是延时去处理消息，我们在发消息都是马上插入到消息队列当中。
• 插入完消息之后，怎么又保证在我们预期的时间里处理消息呢？
  • 需要看一下next()方法。

04.next()方法源码分析

4.1 源码流程分析

• next方法源码如下所示
  • 在 MessageQueue 中消息的读取其实是通过内部的 next() 方法进行的，next() 方法是一个无限循环的方法。博客
  • 如果消息队列中没有消息，则该方法会一直阻塞，
  • 当有新消息来的时候 next() 方法会返回这条消息并将其从单链表中删除。

  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);
  
          synchronized (this) {
              // Try to retrieve the next message.  Return if found.
              final long now = SystemClock.uptimeMillis();
              Message prevMsg = null;
              Message msg = mMessages;
              if (msg != null && msg.target == null) {
                  // Stalled by a barrier.  Find the next asynchronous message in the queue.
                  do {
                      prevMsg = msg;
                      msg = msg.next;
                  } while (msg != null && !msg.isAsynchronous());
              }
              if (msg != null) {
                  if (now < msg.when) {
                      // Next message is not ready.  Set a timeout to wake up when it is ready.
                      nextPollTimeoutMillis = (int) Math.min(msg.when - now, Integer.MAX_VALUE);
                  } else {
                      // Got a message.
                      mBlocked = false;
                      if (prevMsg != null) {
                          prevMsg.next = msg.next;
                      } else {
                          mMessages = msg.next;
                      }
                      msg.next = null;
                      if (DEBUG) Log.v(TAG, "Returning message: " + msg);
                      msg.markInUse();
                      return msg;
                  }
              } else {
                  // No more messages.
                  nextPollTimeoutMillis = -1;
              }
  
              // Process the quit message now that all pending messages have been handled.
              if (mQuitting) {
                  dispose();
                  return null;
              }
  
              // If first time idle, then get the number of idlers to run.
              // Idle handles only run if the queue is empty or if the first message
              // in the queue (possibly a barrier) is due to be handled in the future.
              if (pendingIdleHandlerCount < 0
                      && (mMessages == null || now < mMessages.when)) {
                  pendingIdleHandlerCount = mIdleHandlers.size();
              }
              if (pendingIdleHandlerCount <= 0) {
                  // No idle handlers to run.  Loop and wait some more.
                  mBlocked = true;
                  continue;
              }
  
              if (mPendingIdleHandlers == null) {
                  mPendingIdleHandlers = new IdleHandler[Math.max(pendingIdleHandlerCount, 4)];
              }
              mPendingIdleHandlers = mIdleHandlers.toArray(mPendingIdleHandlers);
          }
  
          // Run the idle handlers.
          // We only ever reach this code block during the first iteration.
          for (int i = 0; i < pendingIdleHandlerCount; i++) {
              final IdleHandler idler = mPendingIdleHandlers[i];
              mPendingIdleHandlers[i] = null; // release the reference to the handler
  
              boolean keep = false;
              try {
                  keep = idler.queueIdle();
              } catch (Throwable t) {
                  Log.wtf(TAG, "IdleHandler threw exception", t);
              }
  
              if (!keep) {
                  synchronized (this) {
                      mIdleHandlers.remove(idler);
                  }
              }
          }
  
          // Reset the idle handler count to 0 so we do not run them again.
          pendingIdleHandlerCount = 0;
  
          // While calling an idle handler, a new message could have been delivered
          // so go back and look again for a pending message without waiting.
          nextPollTimeoutMillis = 0;
      }
  }

4.2 处理延迟消息逻辑

• 如何处理延迟消息的获取
  • msg != null 我们看下这部分，如果当前时间小于头部时间（消息队列是按时间顺序排列的），那就更新等待时间nextPollTimeoutMillis，等下次再做比较。
  • 如果时间到了，就取这个消息并返回。如果没有消息，nextPollTimeoutMillis被赋为-1，这个循环又执行到nativePollOnce继续阻塞。
• nativePollOnce最终执行到的函数
  • Looper.cpp:: pollInner。可以看到是由 epoll_wait来阻塞的。

  int Looper::pollInner(int timeoutMillis) {
         ...
      // Poll.
      int result = POLL_WAKE;
      mResponses.clear();
      mResponseIndex = 0;
      mPolling = true;//即将处于idle状态
  
      struct epoll_event eventItems[EPOLL_MAX_EVENTS];
      int eventCount = epoll_wait(mEpollFd.get(), eventItems, EPOLL_MAX_EVENTS, timeoutMillis);//等待事件发生或者超时，在nativeWake()方法，向管道写端写入字符，则该方法会返回；
      ...
      return result;
  }

05.延迟发送消息队列分析

• 消息队列如下所示
  • 
    image
• 通过上述片段我们了解到
  • 消息队列是按消息触发时间排序的
  • 设置epoll_wait的超时时间，使其在特定时间唤醒，这里我们先计算当前时间和触发时间有多长，这个差值作为epoll_wait的超时时间。
  • epoll_wait超时的时候就是消息触发的时候了，就不会继续堵塞，继续往下执行，这个线程就会被唤醒，去执行消息处理

其他介绍

01.关于博客汇总链接

• 1.技术博客汇总
• 2.开源项目汇总
• 3.生活博客汇总
• 4.喜马拉雅音频汇总
• 5.其他汇总

02.关于我的博客

• github：https://github.com/yangchong211
• 知乎：https://www.zhihu.com/people/yczbj/activities
• 简书：http://www.jianshu.com/u/b7b2c6ed9284
• csdn：http://my.csdn.net/m0_37700275
• 喜马拉雅听书：http://www.ximalaya.com/zhubo/71989305/
• 开源中国：https://my.oschina.net/zbj1618/blog
• 泡在网上的日子：http://www.jcodecraeer.com/member/content_list.php?channelid=1
• 邮箱：yangchong211@163.com
• 阿里云博客：https://yq.aliyun.com/users/article?spm=5176.100- 239.headeruserinfo.3.dT4bcV
• segmentfault头条：https://segmentfault.com/u/xiangjianyu/articles
• 掘金：https://juejin.im/user/5939433efe88c2006afa0c6e