我们知道Java层 有Handler、Looper、MessageQueue的 铁三角模型，Android的消息循环真的就只有这些么?
Android 2.2 之前确实只有Java 层的MessageQueue，但是Android 2.2之后，MessageQueue的核心功能下移到了Native层，导致现在的MessageQueue除了支持Java层的Message派发外,还新增了支持Native Message派发,以及处理来自所监控的文件句柄的事件。
Java 层的MessageQueue创建时,在Native层会创建一个NativeMessageQueue对象,
NativeMessageQueue 持有一个Looper（C++)。
一、sendMessage发送消息
Native层 通过Looper 来发送消息
Native MessageQueue 通过Looper 来发送Message
1.1、Native层 Looper持有一个mMessageEnvelopes 来存储待处理的Message
mMessageEnvelopes中的MessageEnvelope也是按照触发时间,从小到大排序的，排在前面的优先处理。
Vector<MessageEnvelope> mMessageEnvelopes; // guarded by mLock
1.2、Looper发送消息,最终会进入sendMessageAtTime()
Native sendMessageAtTime()方法做的事情和Java层很相似:
- 将新Message 添加到消息队列(mMessageEnvelopes)的合适的位置。
- 新Message插入 队首时,需要立即唤醒队列wake()
void Looper::sendMessageAtTime(nsecs_t uptime, const sp<MessageHandler>& handler,
const Message& message) {
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);
if (mSendingMessage) {
return;
}
} // release lock
// Wake the poll loop only when we enqueue a new message at the head.
if (i == 0) {
wake();
}
}
由上面可知
Java层和Natvie的MessageQueue，有各自维护的独立的消息队列,可以互不干扰的向消息队列中插入Message.
二、MessageQueue处理消息
在Android的MessqgeQueue模型中,Java和Native层 sendMessage是各自独立的,消费队里中的消息却是绑定在一起的。
2.1、MessageQueue.nativePollOnce()
Java层消息循环中,调用MessageQueue.next(),获取下一条待处理的消息。
next()方法中会调用nativePollOnce()方法。
对于Java层nativePollOnce() 会阻塞当前线程
- 当队列没有消息时 会阻塞
- 当队列头部消息 未到达触发时间(msg.when > now），阻塞一个固定的时间，超时后返回。
对于Natvie层nativePollOnce()的调用 是Native层消息被处理的唯一时机。
int Looper::pollInner(int timeoutMillis) {
struct epoll_event eventItems[EPOLL_MAX_EVENTS];
//(1)epoll_wait()返回，阻塞结束
int eventCount = epoll_wait(mEpollFd.get(), eventItems, EPOLL_MAX_EVENTS, timeoutMillis);
if (eventCount < 0) {
if (errno == EINTR) {
goto Done;
}
result = POLL_ERROR;
goto Done;
}
// Check for poll timeout.
if (eventCount == 0) {
result = POLL_TIMEOUT;
goto Done;
}
for (int i = 0; i < eventCount; i++) {
int fd = eventItems[i].data.fd;
uint32_t epollEvents = eventItems[i].events;
if (fd == mWakeEventFd.get()) {
if (epollEvents & EPOLLIN) {
awoken();
} else {
ALOGW("Ignoring unexpected epoll events 0x%x on wake event fd.", epollEvents);
}
} 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);
}
}
}
Done: ;
// (2) 处理Native层的消息
mNextMessageUptime = LLONG_MAX;
while (mMessageEnvelopes.size() != 0) {
nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
const MessageEnvelope& messageEnvelope = mMessageEnvelopes.itemAt(0);
if (messageEnvelope.uptime <= now) {
.
{ // obtain handler
sp<MessageHandler> handler = messageEnvelope.handler;
Message message = messageEnvelope.message;
mMessageEnvelopes.removeAt(0);
mSendingMessage = true;
mLock.unlock();
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++) {
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;
response.request.callback.clear();
result = POLL_CALLBACK;
}
}
return result;
}
- (1)epoll_wait()返回,表示阻塞结束
- (2) Done: 代码块 遍历消息队列,处理到期的message
nativePollOnce() 阻塞时长
阻塞时长 最终取决于epoll_wait 传入的超时时间
int eventCount = epoll_wait(mEpollFd.get(), eventItems, EPOLL_MAX_EVENTS, timeoutMillis);
计算过程如下:
int Looper::pollInner(int timeoutMillis) {
if (timeoutMillis != 0 && mNextMessageUptime != LLONG_MAX) {
nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);
//messageTimeoutMillis 表示Native层下一条消息的到期时间
int messageTimeoutMillis = toMillisecondTimeoutDelay(now, mNextMessageUptime);
if (messageTimeoutMillis >= 0
&& (timeoutMillis < 0 || messageTimeoutMillis < timeoutMillis)) {
timeoutMillis = messageTimeoutMillis;
}
}
- Java层消息队列未空时,nativePollOnce(-1)，timeoutMillis 取决于Native消息队里最新一条消息的到期时间
- Java层 nativePollOnce(0) 时,timeoutMillis = 0,不阻塞
- Java层 传递timeoutMillis为最新一条Message的超时时间时,最终的超时时间timeoutMillis 取Java和Native层最新消息到期执行时间的最小值。
epoll_wait的唤醒
epoll_wait的唤醒分为下面几种情况:
- epoll_wait 超时返回
- epoll_wait 出错
- Java层sendMessage调用natvieWake()
- Native层sendMessage 调用wake()
- 其他监听的Fd事件
三、Android MessageQueue模型小结
- Java层和Native层MessageQueue有各自的消息队列,sendMessage是彼此独立的。
- Java层和Native层MessageQueue处理消息是绑定在一起的，纽带就是nativePollOnce()
- nativePollOnce()是Native层消息得以处理的唯一路径
- natvie层的Mesage消息要先于Java层的Mesage被处理。Native层的消息处理完成之后,Java层的消息才会被处理。
四、Native MessageQueue 可以处理哪些消息呢?
Looper/Handler的模型主要是依靠不断的唤醒epoll来实现的。主要的手段是唤醒 mWakeFd, 或者是epoll的timeout。
timeout手段主要是针对 sendMessageDelayed()这样的函数。
epoll 的唤醒除了 mWakeFd的唤醒和epoll的timeout外，还支持监听其它事件。主要是通过Looper::addFd来实现的。
int Looper::addFd(int fd, int ident, int events, Looper_callbackFunc callback, void* data) {
return addFd(fd, ident, events, callback ? new SimpleLooperCallback(callback) : nullptr, data);
}
int Looper::addFd(int fd, int ident, int events, const sp<LooperCallback>& callback, void* data) {
{ // 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.get(), EPOLL_CTL_ADD, fd, &eventItem);
if (epollResult < 0) {
ALOGE("Error adding epoll events for fd %d: %s", fd, strerror(errno));
return -1;
}
mRequests.add(fd, request);
} else {
int epollResult = epoll_ctl(mEpollFd.get(), EPOLL_CTL_MOD, fd, &eventItem);
if (epollResult < 0) {
if (errno == ENOENT) {
epollResult = epoll_ctl(mEpollFd.get(), EPOLL_CTL_ADD, fd, &eventItem);
if (epollResult < 0) {
ALOGE("Error modifying or adding epoll events for fd %d: %s",
fd, strerror(errno));
return -1;
}
scheduleEpollRebuildLocked();
} else {
ALOGE("Error modifying epoll events for fd %d: %s", fd, strerror(errno));
return -1;
}
}
mRequests.replaceValueAt(requestIndex, request);
}
} // release lock
return 1;
}
从上面的代码可以看出addFd就是创建一个Request，并添加到epoll监听池中，并插入到mRequests中。
epoll_wait()函数等待消息,当接收到消息或者消息发生超时后调用返回。
对返回的events 进行遍历，如果对应的fd为唤醒专用的mWakeEventId,执行awoken()函数清空管道,这个事件的作用只是为了唤醒Looper对新消息进行处理。
如果不是mWakeEventFd，说明为我们之前通过addFd()函数添加的自定义fd，我们需要对这个event进行处理，处理函数为pushResponse()：
ssize_t requestIndex = mRequests.indexOfKey(fd);
pushResponse(events, mRequests.valueAt(requestIndex));
这里将request对象包装成了一个response，然后存入了mResponses中等待后面的处理。
void Looper::pushResponse(int events, const Request& request) {
Response response;
response.events = events;
response.request = request;
mResponses.push(response);
}
然后是对response的处理操作,主要是如下调用
int callbackResult = response.request.callback->handleEvent(fd, events, data);
4.1、举个栗子
Android的屏幕触摸事件以及按键事件 就是通过Looper::addFd()的方法,经由Native的MessageQueue 进行事件分发的。
void NativeInputEventReceiver::setFdEvents(int events) {
if (mFdEvents != events) {
mFdEvents = events;
int fd = mInputConsumer.getChannel()->getFd();
if (events) {
mMessageQueue->getLooper()->addFd(fd, 0, events, this, nullptr);
} else {
mMessageQueue->getLooper()->removeFd(fd);
}
}
}
int fd = mInputConsumer.getChannel()->getFd() 是一个socket的fd文件描述符。
NativeInputEventReceiver 通过addFd，将自己作为监听回调注册到了Looper中。
status_t InputChannel::sendMessage(const InputMessage* msg) {
size_t msgLength = msg->size();
ssize_t nWrite;
do {
nWrite = ::send(mFd, msg, msgLength, MSG_DONTWAIT | MSG_NOSIGNAL);
} while (nWrite == -1 && errno == EINTR);
...
return OK;
}
当原始事件通过建立好的InputChannel的sendMessage()函数发送触摸事件时，会直接调用send()函数向fd表示的socket中写入数据，同时在另一边的epoll_wait()调用就会因socket另一端fd数据的到来而唤醒，并通过fd找到注册好的request，进而调用request中的NativeInputEventReceiver的handleEvent()方法，参数就是我们接收到的事件信息与数据。
4.2、彩蛋
BlockCanary 缺点分析一文中指出,MessageQueue.next()阻塞 导致的ANR,BlockCanary是捕捉不到的。
Message msg = queue.next(); // might block
用举出实例,在Activity中重写dispatchTouchEvent和dispatchKeyEvent，模拟耗时操作，弹出ANR告警，但BlockCanary没有任何反应。
@Override
public boolean dispatchTouchEvent(MotionEvent event){
Log.d("feifei","dispatchTouchEvent A ");
try {
Thread.sleep(20000);
} catch (InterruptedException e) {
e.printStackTrace();
}
Log.d("feifei","dispatchTouchEvent B ");
return super.dispatchTouchEvent(event);
}
我试验了一次,BlockCanary缺失没有捕获ANR。
问题来了
为什么dispatchTouchEvent执行耗时操作,MessageQueue.next()就会被阻塞呢?
经过本文对Native层的MessageQueue的运行机制梳理,可以知道:
- 触摸事件,是Native层的MessageQueue,通过NativeInputEventReceiver进行事件分发的。
- NativePollOnce()进入native层调用epoll_wait()阻塞线程,阻塞解除后,会先对Native层的消息进行分发处理,处理完成之后,才会返回Java层。
综合上面两点:dispatchTouchEvent中执行耗时操作,会直接导致Native层分发消息耗时,进而导致Java层MessageQueue.next()长时间阻塞，触发ANR。
五、参考文章
https://www.jianshu.com/p/48cf21ad637b
