美文网首页
MessageQueue之分析native方法

MessageQueue之分析native方法

作者: Lemon666 | 来源:发表于2020-03-26 15:59 被阅读0次

    了解过Handler的应该知道,Looper获取Message是从MessageQueue的next()方法获取,在next()方法里面如果下一个Message执行的时间还未到,则会调用native方法nativePollOnce()阻塞,直到下一个Message执行的时间

    MessageQueue可以看到有这么几个native方法

    private native static long nativeInit();
    private native static void nativeDestroy(long ptr);
    private native void nativePollOnce(long ptr, int timeoutMillis); /*non-static for callbacks*/
    private native static void nativeWake(long ptr);
    private native static boolean nativeIsPolling(long ptr);
    private native static void nativeSetFileDescriptorEvents(longptr, int fd, int events);
    

    这里主要分析nativeInit,nativePollOnce和nativeWake
    分别是初始化,阻塞和唤醒

    分析源码之前,需要先了解一下linux的epoll

    //调用epoll_create建立一个epoll对象(在epoll文件系统中给这个句柄分配资源)
    int epoll_create(int size);  
    //将被监听的描述符添加到epoll句柄或从epool句柄中删除或者对监听事件进行修改
    int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); 
    //调用epoll_wait收集发生事件的连接
    int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);
    

    一般使用方法

    //假设需要监听的文件为fd
    //省略获取文件描述符的逻辑
    int fd;
    int size = 256;
    //先创建epoll对象,size用来告诉内核监听的数量
    int epfd = epoll_create(size);
    //添加文件描述符到epfd
    struct epoll_event eventItem,eventItems[size];
    eventItem.events = EPOLLIN;//监听事件
    eventItem.data.fd = fd;//监听的文件描述符
    epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&eventItem)
    //开始阻塞监听
    int timeoutMillis = -1;//这里设置为-1,直到有内容可以才返回
    int eventCount = epoll_wait(epfd, eventItems, size, timeoutMillis);
    //如果fd文件描述符号有内容可读,则返回eventCount,表示有多少个fd文件描述符可以读,我们这里只监听1个fd
    for(int i=0;i<eventCount;i++){
        eventItems[i];//这里拿到变动过的fd就可以进行相应操作
    }
    

    以上是epoll基本使用的过程

    MessageQueue构造方法

    //调用nativitInit(),创建NativeMessageQueue
    MessageQueue(boolean quitAllowed) {
        mQuitAllowed = quitAllowed;
        mPtr = nativeInit();
    }
    
    static jlong android_os_MessageQueue_nativeInit(JNIEnv* env, jclass clazz) {
        //构造NativeMessageQueue
        NativeMessageQueue* nativeMessageQueue = new NativeMessageQueue();
        if (!nativeMessageQueue) {
            jniThrowRuntimeException(env, "Unable to allocate native queue");
        nativeMessageQueue->incStrong(env);
        //返回NativeMessageQueue地址
        return reinterpret_cast<jlong>(nativeMessageQueue);
    }
    //NativeMessageQueue构造函数
    NativeMessageQueue::NativeMessageQueue() :
            mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {
        //获取线程变量Looper,也就是线程唯一,和Java的ThreadLocal一样
        mLooper = Looper::getForThread();
        if (mLooper == NULL) {
            //如何为Null,则初始化然后设置到线程变量
            mLooper = new Looper(false);
            Looper::setForThread(mLooper);
        }
    }
    

    Looper构造方法

    Looper::Looper(bool allowNonCallbacks) :
            mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),
            mPolling(false), mEpollFd(-1), mEpollRebuildRequired(false),
            mNextRequestSeq(0), mResponseIndex(0), mNextMessageUptime(LLONG_MAX) {
        //eventfd是Linux2.6提供的一种系统调用,它可以用来实现事件通。
        //eventfd包含一个由内核维护的64位无符号整型计数器,
        //创建eventfd时会返回一个文件描述符,进程可以通过对
        //这个文件描述符进行read/write来读取/改变计数器的值,从而实现进程间通信。
        //返回mWakeEventFd就是我们需要监听的文件描述符
        mWakeEventFd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
        LOG_ALWAYS_FATAL_IF(mWakeEventFd < 0, "Could not make wake event fd: %s",
                            strerror(errno));
        AutoMutex _l(mLock);
        rebuildEpollLocked();
    }
    //rebuildEpollLocked()创建epoll监听文件描述符
    void Looper::rebuildEpollLocked() {
        // Close old epoll instance if we have one.
        if (mEpollFd >= 0) {
            //如果已经创建过epoll,则关闭重新创建
            close(mEpollFd);
        }
        // Allocate the new epoll instance and register the wake pipe.
        创建一个epoll,返回epoll的文件描述符
        mEpollFd = epoll_create(EPOLL_SIZE_HINT);
      
        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;
        //把mWakeEventFd,也就是eventFd创建文件描述加入epoll监听
        int result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeEventFd, & eventItem);
        //省略部分代码...
    }
    
    

    Android会在创建MessageQueue的时候,创建NativeMessageQueue,在NativeMessageQueue,创建Looper,并且创建epoll去监听eventFd

    nativePollOnce()

    Message next() {
        //省略部分代码...
        for (;;) {
            //省略部分代码...
            //阻塞线程直到下一个Message的执行时间
            //nextPollTimeoutMillis
            // -1: 一直阻塞,直到被唤醒
            //  0: 立即返回,
            // >0: 阻塞固定时间后返回
            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) {
                    if (now < msg.when) {
                        //如果下个Message需要执行的时间大于当前时间,则计算获取时间差
                        // 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;
                        msg.markInUse();
                        //下个Message可以立即执行,则返回该Message
                        return msg;
                    }
                } else {
                    // No more messages.
                    // 消息队列为空,则一直阻塞直到有新的Message进来
                    nextPollTimeoutMillis = -1;
                }
                //省略部分代码...
            }
            //省略部分代码...
            nextPollTimeoutMillis = 0;
        }
    }
    
    //MessageQueue.java
    //在next()方法调用nativePollOnce,传入地址和阻塞时间
    nativePollOnce(ptr, nextPollTimeoutMillis);
    
    ///frameworks/base/core/jni/android_os_MessageQueue.cpp
    static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj,
            jlong ptr, jint timeoutMillis) {
        //根据MessageQueue传过来的ptr获取到NativeMessageQueue对象
        NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
        //调用pollOnce,传入阻塞的时间参数
        nativeMessageQueue->pollOnce(env, obj, timeoutMillis);
    }
    
    //NativeMessageQueue#pollOnce
    void NativeMessageQueue::pollOnce(JNIEnv* env, jobject pollObj, int timeoutMillis) {
        //省略部分代码...
        //调用了Looper的pollOnce
        mLooper->pollOnce(timeoutMillis);
        //省略部分代码...
    }
    
    //system/core/include/utils/Looper.h
    inline int pollOnce(int timeoutMillis) {
        return pollOnce(timeoutMillis, NULL, NULL, NULL);
    }
    int pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData);
    //最终调用pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData),这个方法在Looper.cpp实现
    
    //system/core/libutils/Looper.cpp
    int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) {
        int result = 0;
        for (;;) {
            // 省略部分代码...
            if (result != 0) {
                // 省略部分代码...
                return result;
            }
            result = pollInner(timeoutMillis);
        }
    }
    

    再看pollInner(timeoutMillis)

    int Looper::pollInner(int timeoutMillis) {
        // 省略部分代码...
        int result = POLL_WAKE;
        // 创建epoll_event结构体数组,存放返回的epoll_event
        struct epoll_event eventItems[EPOLL_MAX_EVENTS];
        // 阻塞等待
        int eventCount = epoll_wait(mEpollFd, eventItems, LL_MAX_EVENTS, timeoutMillis);
        // 省略部分代码...
        // 循环遍历处理epoll对应文件描述符IO事件
        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) {
                    //读取描述符号eventfd数据
                    awoken();
                } 
                // ...
            } else {
                // 省略部分代码...
            }
        }
        // 省略部分代码...
        return result;
    }
    

    awoken()

    void Looper::awoken() {
        uint64_t counter;
        TEMP_FAILURE_RETRY(read(mWakeEventFd, &counter, eof(uint64_t)));
    }
    

    如果中途在MessageQueue插入Message

    //Handler.java
    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);
    }
    //再走enqueueMessage
    private boolean enqueueMessage(MessageQueue queue, Message msg, long uptimeMillis) {
        msg.target = this;
        if (mAsynchronous) {
            msg.setAsynchronous(true);
        }
        //最后通过MessageQueue.enqueueMessage插入Message
        return queue.enqueueMessage(msg, uptimeMillis);
    }
    
    //MessageQueue.java
    boolean enqueueMessage(Message msg, long when) {
        //省略部分代码
        synchronized (this) {
             //省略部分代码
            Message p = mMessages;
            boolean needWake;
            if (p == null || when == 0 || when < p.when) {
                //...
                //消息队列为空
                //执行时间为0,0肯定小于当前时间
                //执行时间小于最队列第一个Messag的执行时间
                //如果当前是阻塞状态则需要唤醒
                needWake = mBlocked;
            } else {
                //...
                // 存在同步屏障,并且插入的消息是异步消息
                needWake = mBlocked && p.target == null && msg.isAsynchronous(); 
                //...
                //插入消息队列
                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;
    }
    

    跟踪nativeWake

    ///base/core/jni/android_os_MessageQueue.cpp
    static void android_os_MessageQueue_nativeWake(JNIEnv* env, jclass clazz, jlong ptr) {
        NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
        nativeMessageQueue->wake();
    }
    void NativeMessageQueue::wake() {
        //最终调用Looper.wake方法
        mLooper->wake();
    }
    
    
    void Looper::wake() {
        //往mWakeEventFd文件描述符写数据
        uint64_t inc = 1;
        ssize_t nWrite = TEMP_FAILURE_RETRY(write(mWakeEventFd, &inc, sizeof(uint64_t)));
                LOG_ALWAYS_FATAL("Could not write wake signal to fd %d: %s",
                        mWakeEventFd, strerror(errno));
            }
        }
    }
    

    相关文章

      网友评论

          本文标题:MessageQueue之分析native方法

          本文链接:https://www.haomeiwen.com/subject/cuucuhtx.html