Native 层的MessageQueue简析

作者: feifei_fly | 来源:发表于2021-01-13 20:35 被阅读0次

Native 层的MessageQueue简析
Android通信方式篇（三）-消息机制(Native层)
Android消息机制（native层）
MessageQueue, Looper源码分析（Native层
react native资料
Android消息机制Handler(Native层)
理解 MessageQueue
H5混合开发
RN精进笔记（四）通讯原理篇
JNI 原理

image

我们知道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模型小结

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() shi 的一个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

https://www.cnblogs.com/angeldevil/p/3340644.html

https://www.viseator.com/2017/10/07/android_view_event_2/

Native 层的MessageQueue简析
Java层MessageQueue分析[https://www.jianshu.com/p/8b30773c91a...
Android通信方式篇（三）-消息机制(Native层)
在前面介绍的Java层中，我们看到了MessageQueue有若干native方法，想必肯定与native层有关，...
Android消息机制（native层）
前言上一篇文章：Android消息机制java层 MessageQueue中有多个native方法，Messae...
MessageQueue, Looper源码分析（Native层
本源码分析基于Android8.0 源码目录回顾在上一篇文章中，我们讲解了Handler，Looper，M...
react native资料
React Native 学习资源精选仓库(汇聚知识，分享精华) React Native IOS集成与原理简析 ...
Android消息机制Handler(Native层)
一、概论在 java 层的消息处理机制，其中MessageQueue类里面涉及到多个native方法，除了Mes...
理解 MessageQueue
Android的MessageQueue经历过一次大的变化，在2.0的时候，还没有native这一层。但到了2.3...
H5混合开发
1. 混合方案简析 Hybrid App，俗称混合应用，即混合了 Native技术与 Web技术进行开发的移...
RN精进笔记（四）通讯原理篇
React-Native通讯原理设置SPY_MODE标志为true //MessageQueue.js，需要处于...
JNI 原理
我们都知道JNI结构是 Java 层 -> JNI -> Native 层，以此实现Java 层和Native层...