Android 线程间通讯机制

Posted 2022-12-11 白嫩豆腐

前言

之前研究Fragment遇到进程间通讯的一些东东，趁着最近有空，就在重新研究一下android相关的代码。并且这些代码确实非常简单。之前研究过，但是遇到一些问题就没深究，这次我们就彻底搞懂他吧

正文

最简单的使用当然是在activity的oncreat中直接使用：

Handler handler = new Handler();
handler.post(new runable
	Log.e("mainactivity","我是handler post的log");
)

但是我们的主线程初始化时候帮我们干了一大堆事情，研究不够典型，我们直接看新线程如何通讯，直接用HandlerThread来作为源码研究的调用开端

myHandlerThread = new HandlerThread( "handler-thread") ;
        //开启一个线程
myHandlerThread.start();
        //在这个线程中创建一个handler对象
handler = new Handler(myHandlerThread.getLooper())
handler.post(new runable
	Log.e("mainactivity","我是handler post的log");
)

相信你第一眼目光就集中在run()函数中了，内容如下

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

这个才是真的进程间通讯的核心,这里我们就一点一点阅读：

    /** Initialize the current thread as a looper.
      * This gives you a chance to create handlers that then reference
      * this looper, before actually starting the loop. Be sure to call
      * @link #loop() after calling this method, and end it by calling
      * @link #quit().
      */
    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));
    

注释解释的很明白。就是初始化一个Looper。其实可能有疑惑的地方是sThreadLocal.set()这是维持在当前线程的一个变量，可以在任何线程中，调用myLooper。得到唯一的一个实例。暂时对我们问题空间没有影响。我们只要关心new Looper(quitAllowed)即可

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

有些出乎意料，太简单了吧！可是短小的都是精华，脚下就是万丈深渊

    MessageQueue(boolean quitAllowed) 
        mQuitAllowed = quitAllowed;
        mPtr = nativeInit();
    
        private native static long nativeInit();

emmm忽然进入native层了，有些让人望而却步。不过我们看过以前大佬的书和博客，还是很容易定为到源码的位置aosp\\frameworks\\base\\core\\jni\\android_os_MessageQueue.cpp

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);
    

貌似和java层差不多，都是初始化了一个Looper。所有的等待和定时其实都是通过c层的Looper来实现的。我们这里就一一分析

Looper::Looper(bool allowNonCallbacks)
    
    //初始化一个文件，作为中断的一端
    mWakeEventFd.reset(eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC));
    LOG_ALWAYS_FATAL_IF(mWakeEventFd.get() < 0, "Could not make wake event fd: %s", strerror(errno));

    AutoMutex _l(mLock);
    //关键部分
    rebuildEpollLocked();

这里所以的东西都是通过epoll来实现的休眠定时唤醒等操作。epoll大概原理是类似一个回调，但是属于系统层的。epoll可以监听一个文件是否变化，变化时候，可以回调函数，若文件不变，则线程会一直休眠。我们来看下相关代码

void Looper::rebuildEpollLocked() 

    // Allocate the new epoll instance and register the wake pipe.
    mEpollFd.reset(epoll_create1(EPOLL_CLOEXEC));

    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.get();
    int result = epoll_ctl(mEpollFd.get(), EPOLL_CTL_ADD, mWakeEventFd.get(), &eventItem);

//这个是为了系统层的一些东西准备的，这都没被调用所以可以注释掉
/**
    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.get(), EPOLL_CTL_ADD, request.fd, &eventItem);
        if (epollResult < 0) 
            ALOGE("Error adding epoll events for fd %d while rebuilding epoll set: %s",
                  request.fd, strerror(errno));
        
    
**/

通过epoll_create1初始化。epoll_ctl注册回调监听.所有的事情都干完了。
下面进入真的loop函数了

/**
     * Run the message queue in this thread. Be sure to call
     * @link #quit() to end the loop.
     */
    public static void loop() 
        final Looper me = myLooper();
       
        final MessageQueue queue = me.mQueue;

        for (;;) 
            Message msg = queue.next(); // might block
            if (msg == null) 
                // No message indicates that the message queue is quitting.
                return;
            

            try 
                msg.target.dispatchMessage(msg);
             finally 
            
            msg.recycleUnchecked();
        
    

很长的代码，但是貌似其他的内容都是线程安全和性能分析的trace的。所以我们就分析一下这些，其实看名字就容易理解。looper是一个终极循环，loop函数是终极循环的循环体。messagequeue是一个消息队列如果看注释。在queue的next方法中，他注释是可能阻塞，也就是没消息时候，就阻塞一下。整个逻辑就更清晰了。
我们还是看我们的queue的next方法，这里阻塞是如何实现的：

    Message next() 
   
        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;
                

             
             
            // 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;
        
    

主要函数还是这个nativePollOnce(ptr, ia)。ptr是cpp底层的messagequeue保存一个looper。其实最终就是调用了lopper的pollonce。我们直接看最终代码

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 (outFd != NULL) *outFd = fd;
                if (outEvents != NULL) *outEvents = events;
                if (outData != NULL) *outData = data;
                return ident;
            
        

        if (result != 0) 
            if (outFd != NULL) *outFd = 0;
            if (outEvents != NULL) *outEvents = 0;
            if (outData != NULL) *outData = NULL;
            return result;
        

        result = pollInner(timeoutMillis);
    

两个循环让人摸不着头脑，但是我们很容易确定mResponses里面没内容。所以就是等待pollInner函数返回部为0的值。继续追pollInner:

int Looper::pollInner(int timeoutMillis) 

    // 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;
        
    

    // Poll.
    int result = POLL_WAKE;
    mResponses.clear();
    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);

    // No longer idling.
    mPolling = false;

    // Acquire lock.
    mLock.lock();

    // Handle all events.

    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);
            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: ;

    // Invoke pending message callbacks.
    mNextMessageUptime = LLONG_MAX;
    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();

                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;
            // 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;

真个流程比较复杂，但是真的有用的地方只有那个int eventCount = epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis);他是阻塞线程的核心，后面的就是处理一些屏幕点击事件唤醒线程的东东，完全可以无视，所以就返回了int result = POLL_WAKE
wait才是真正的阻塞，初始化的时候，timeoutMillis都为0，所以是不会阻塞的，直接返回到Looper中的loop函数中。
loop函数中。再次进入cpp层，timeoutMillis变成了-1.然后就可以睡眠了，等待监听的文件有写入操作，继续进入loop函数的无限循环。
下面我们分析handlerThread的post发送消息。
这里很容易进入核心的函数调用到了messageQueue中的enqueueMessage

    boolean enqueueMessage(Message msg, long when) 
        
        synchronized (this) 
        
            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;
    

其实这个代码会比较简单，就是判断消息队列mMessages是否为空，为空就吧自己当成消息队列的开头，然后通过一个for循环，把当前的消息插入合适的位置，ps只有当当前消息的when小于消息队列中的最后一个，或者消息队列后面已经没内容的时候，就把消息