Android消息机制,从Java层到Native层剖析

Android消息机制,从Java层到Native层剖析

由Handler、MessageQueue、Looper构成的线程消息通信机制在Android开发中非常常用,不过大部分人都只粗浅地看了Java层的实现,对其中的细节不甚了了,这篇博文将研究Android消息机制从Java层到Native层的实现。

消息机制由于更贴近抽象设计,所以整个结构更简单,只包含了消息的产生、分发,不像Input子系统那样还有归类、过滤等环节。整体的结构如下图:

Android Java层消息机制

消息的产生

在Java层中消息的产生都来源于用户创建的Message对象,经过封装的Runnable对象,或调用obtainMessage从Message Pool中获得,Message Pool指的是Message类内的Message循环队列,队头是静态的Message对象sPool,该队列最大容纳MAX_POOL_SIZE(50)个Message:

MessagePool对Message的复用节省了不断创建Message带来的开销,如果当前50个Message都已经被用过,由于MessagePool是循环队列,则会回到队头并请空该Message,向下复用。

BlockingRunnable

看Java层Handler的源码的时候发现了一个奇怪的东西:BlockingRunnable,基本上没有用过的东西,也没看别人讲过,于是我就来钻研一下吧:

private static final class BlockingRunnable implements Runnable {
    private final Runnable mTask;
    private boolean mDone;

    public BlockingRunnable(Runnable task) {
        mTask = task;
    }

    @Override
    public void run() {
        try {
            mTask.run();
        } finally {
            synchronized (this) {
                mDone = true;
                notifyAll();
            }
        }
    }

    public boolean postAndWait(Handler handler, long timeout) {
        if (!handler.post(this)) {
            return false;
        }

        synchronized (this) {
            if (timeout > 0) {
                final long expirationTime = SystemClock.uptimeMillis() + timeout;
                while (!mDone) {
                    long delay = expirationTime - SystemClock.uptimeMillis();
                    if (delay <= 0) {
                        return false; // timeout
                    }
                    try {
                        wait(delay);
                    } catch (InterruptedException ex) {
                    }
                }
            } else {
                while (!mDone) {
                    try {
                        wait();
                    } catch (InterruptedException ex) {
                    }
                }
            }
        }
        return true;
    }
}

我们可以看到,BlockingRunnable是一个“包裹”构造方法中传入的Runnable的Runnable,调用BlockingRunnable的postAndWait会做以下事情:

  1. 如果投递BlockingRunnable失败,返回false
  2. 锁住投递BlockingRunnable的线程
  3. 如果timeout大于0,计算参数Runnable的到期时间,只要参数Runnable还没处理完,则一直轮询还剩多少时间,并调用wait(delay)让投递BlockingRunnable的线程继续等待,直参数Runnable处理完(mDone为true)这个过程才结束
  4. 如果timeout小于等于0,而且参数Runnable还没处理完,则一直等待直到参数Runnable处理完(mDone为true)

这个东西的说明书和使用风险可以在runWithScissors方法的注释里看到,我在这里就不当翻译工了。

消息的投递和处理

得到Message后,就会通过Handler的sendMessageAtTime调用MessageQueue的enqueueMessage将Message投递到MessageQueue中,在往下学习之前必须先了解Handler的创建,因为后面的知识和它有关联。

Handler的创建和初始化

其实Handler的初始化没什么好看的,就是保存Callback、mLooper的MessageQueue的引用,以及声明Handler是否异步投递所有Message。但是里面有一个内存泄露的检查,可以学习一下,就是如果打开了FIND_POTENTIAL_LEAKS,就会进行内存泄露的检查,它会做以下事情:

  1. 获取当前Handler类
  2. 如果Handler是匿名内部类,或成员类,或局部类,且Handler的修饰符不是static
  3. 那么就会打出log提示可能会发生内存泄露
public Handler(Callback callback, boolean async) {
    if (FIND_POTENTIAL_LEAKS) {
        final Class<? extends Handler> klass = getClass();
        if ((klass.isAnonymousClass() || klass.isMemberClass() || klass.isLocalClass()) &&
                (klass.getModifiers() & Modifier.STATIC) == 0) {
            Log.w(TAG, "The following Handler class should be static or leaks might occur: " +
                klass.getCanonicalName());
        }
    }

    mLooper = Looper.myLooper();
    if (mLooper == null) {
        throw new RuntimeException(
            "Can't create handler inside thread that has not called Looper.prepare()");
    }
    mQueue = mLooper.mQueue;
    mCallback = callback;
    mAsynchronous = async;
}

public static @Nullable Looper myLooper() {
    return sThreadLocal.get();
}

既然Handler的创建这么简单,为什么说后面要学习的内容和它相关呢?原因就出在Looper中,我们可以看到Looper是通过sThreadLocal返回的,这个ThreadLocal是什么呢?

ThreadLocal - 维持线程内对象的唯一性

ThreadLocal是一个关于创建线程局部变量的类。

通常情况下,我们创建的变量是可以被任何一个线程访问并修改的。而使用ThreadLocal创建的变量只能被当前线程访问,其他线程则无法访问和修改。它的实现原理如下:

如图所示,ThreadLocalRef其实是同一个ThreadLocal对象的引用,为了不让线看起来很乱我分别用了两个方块表示ThreadLocal对象,但其实是同一个对象。ThreadLocal同时是ThreadA、ThreadB甚至ThreadN内ThreadLocalMap的Key,但取出来的对象时不一样的,因为Map不一样对应的键值对也不一样嘛。

ThreadLocalMap

ThreadLocalMap是仅用于维护ThreadLocal值的自定义HashMap,只在Thread类内使用。为了避免ThreadLocalMap的Key->ThreadLocal在GC时无法被回收,里边的元素都是用WeakReference封装的。ThreadLocalMap除了这点以外,没有什么特别的,就不细讲了。

需要注意的一点是:ThreadLocalMap是可能带来内存泄露的,但root cause不是ThreadLocalMap本身,而是代码质量不够高。首先,由于作为Map的Key的ThreadLocal是弱引用,那么GC时ThreadLocal会被回收,此时Map内存在一对Key为null的键值对,而Value仍然被线程强引用着,那么如果用完ThreadLocal后不主动移除,就会内存泄露了。但事实上,ThreadLocal用完后主动调remove就能规避这个问题,本来也该这样做。

Entry

Entry作为ThreadLocalMap的元素,表示的是一对键值对:ThreadLocal的弱引用为键,将要用ThreadLocal存储的对象为值。

static class Entry extends WeakReference<ThreadLocal> {
    /** The value associated with this ThreadLocal. */
    Object value;

    Entry(ThreadLocal k, Object v) {
        super(k);
        value = v;
    }
}

ThreadLocal总结

换句话说,所谓的不可被其他线程修改的局部变量,表示的是:每个线程中都会维护一个ThreadLocalMap,里边以ThreadLocal为键,对应的局部变量为值,通过键值对来控制访问和数据的一致性,而不是通过锁来控制。

Looper

既然一个线程只有一个Looper,那么Looper里面有什么呢?从源码可以看到,Looper的构造方法是私有的,也就意味着获得Looper对象基本都是单例,这一点和线程Looper的一对一映射关系切合。

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

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的成员变量我们可以知道Looper包含了以下东西:

  • sMainLooper:应用主线程的Looper,创建其他线程的Looper时为null
  • mQueue:Looper关联的MessageQueue
  • mThread:Looper关联的线程
  • sThreadLocal:线程局部变量的Key

从这可以知道,一个线程对应一个Looper,一个Looper对应一个MessageQueue

----------------分割线,接下来回到消息的投递结束的地方----------------

得到Message后,就会通过Handler的sendMessageAtTime调用MessageQueue的enqueueMessage将Message投递到MessageQueue中,在往下学习之前必须先了解Handler的创建,因为后面的知识和它有关联。

现在我们知道Message将要投递到哪里的MessageQueue里了,那么投递过去之后,消息是怎么被处理的呢?这代码很长,而且就是个进入队列的过程,我就不贴了,做了以下事情:

  1. 合法性检查
  2. 标记Message正在使用
  3. 入列
  4. 唤醒native的MessageQueue

在这里有个有意思的概念必须提一下,就是Barrier Message,它表示的是一种栅栏的概念,将它加入MessageQueue可以拦住所有执行时间在它之后的同步Message,异步Message则不受影响,遍历到就会处理,这种状况会持续到把Barrier Message移除。

提示:图里绿色代表Message可以被取出执行,红色表示无法被取出执行

它和Message的根本差别是,他没有target,即:没有处理该Message的Handler,但我们自己将Message的Handler设为null是没法加入MessageQueue的,必须调用postSyncBarrier方法:

boolean enqueueMessage(Message msg, long when) {
    if (msg.target == null) {
        throw new IllegalArgumentException("Message must have a target.");
    }
    ……
}

private int postSyncBarrier(long when) {
    // Enqueue a new sync barrier token.
    // We don't need to wake the queue because the purpose of a barrier is to stall it.
    synchronized (this) {
        final int token = mNextBarrierToken++;
        final Message msg = Message.obtain();
        msg.markInUse();
        msg.when = when;
        msg.arg1 = token;

        Message prev = null;
        Message p = mMessages;
        if (when != 0) {
            while (p != null && p.when <= when) {
                prev = p;
                p = p.next;
            }
        }
        if (prev != null) { // invariant: p == prev.next
            msg.next = p;
            prev.next = msg;
        } else {
            msg.next = p;
            mMessages = msg;
        }
        return token;
    }
}

消息的分发

前面已经知道Message投递后就会到达MessageQueue,接下来就看消息是怎么被遍历处理的。首先要知道的一点是,Looper在调用prepare创建后,是必须调loop()方法的,很多人会问,我平常用的时候没用loop()方法也没问题啊。那是因为你是在主线程用的,主线程在创建Looper的时候已经调用过loop()方法了。

我们创建了其他线程的Looper后,调loop()方法会做以下事情:

  1. 循环获取MessageQueue中的Message
  2. 将Message通过Handler的dispatchMessage方法分发到对应的Handler中
  3. 将Message的信息清空,回收到Message Pool中等待下一次使用
public static void loop() {
    ……

    for (;;) {
        Message msg = queue.next(); // might block

        ……

        try {
            msg.target.dispatchMessage(msg);
        } finally {
            ……
        }

        ……

        msg.recycleUnchecked();
    }
}

在Handler的dispatchMessage中,对Message的处理其实是有优先顺序这个说法的:

  1. 如果Message设置了callback,则将Message交给Message的callback处理
  2. 如果Handler设置了callback,则将Message先交给Handler的callback处理
  3. 否则的话,将Message交给Handler的handleMessage处理
public void dispatchMessage(Message msg) {
    if (msg.callback != null) {
        handleCallback(msg);
    } else {
        if (mCallback != null) {
            if (mCallback.handleMessage(msg)) {
                return;
            }
        }
        handleMessage(msg);
    }
}

对于MessageQueue,它实际表示了Java层和Native层的MessageQueue,Java层的MessageQueue就是mMessages表示的循环队列,Native层的MessageQueue就是mPtr。它的next()方法里做的事情如下:

  1. 调用nativePollOnce让native层的MessageQueue先处理Native层的Message,再处理Java层的Message,这个过程可能阻塞
  2. 如果在按时序遍历MessageQueue的过程中发现了Barrier Message,即handler为空的Message,则跳过它后面的所有同步Message,只处理异步Message
  3. 如果消息是延时消息,计算当前时间和目标时间的差值,休眠这个时间差后再去取这个Message
  4. 如果消息不是延时消息,在Message Pool里标记该Message正在使用,并返回它

Java层Android消息机制的整个过程可以用下图概括:

有钻研过Java层代码的朋友肯定知道,Handler里面还有个用于跨进程Message通信的MessengerImpl,这个东西我就不在这里说了,因为它就是个简单的跨进程通信,和整个Handler、Looper、MessageQueue其实关系不大。

Android Native层消息机制

Android消息机制在Native层其实和Java层很相似,保留了Handler、Looper、MessageQueue的结构。但是Native层Message、Handler、MessageQueue的概念被弱化得很厉害,基本上只是个“空壳”,核心逻辑都在Looper里边了。

其他区别都不大了,只是在实现上有一点不一样,具体的差别就在源码中找答案吧。整体结构图如下:

消息的产生

在Native层中,消息由MessageEnvelope和封装fd(Java层Handler可以添加fd的监听、Native当然也可以)相关信息后得到的epoll_event组成。

fd

对于要被监听的fd的消息,Looper做了以下事情:

  1. 合法性检查
  2. 将相关信息封装到Request中,并初始化为epoll_event
  3. 将该fd以及要监听的epoll_event事件(步骤2转换Request得到)注册到当前Looper的epollFd中
  4. 如果出错,进行出错处理
  5. 更新mRequests
int Looper::addFd(int fd, int ident, int events, const sp<LooperCallback>& callback, void* data) {
    ……

    { // acquire lock
        AutoMutex _l(mLock);

        ……

        struct epoll_event eventItem;
        request.initEventItem(&eventItem);

        ssize_t requestIndex = mRequests.indexOfKey(fd);
        if (requestIndex < 0) {
            int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, fd, & eventItem);
            ……
            mRequests.add(fd, request);
        } else {
            int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_MOD, fd, & eventItem);
            ……
            mRequests.replaceValueAt(requestIndex, request);
        }
    } // release lock
    return 1;
}

MessageEnvelope

MessageEnvelope相对于fd就简单多了,在调用Native层Looper的sendMessage相关函数时会将uptime、MessageHandler、Native层Message封装到MessageEnvelope中,然后插入mMessageEnvelopes中。

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

        ……
    }
    ……
}

消息的投递和处理

前面已经提到了,Java层的MessageQueue处理消息时,会先调用Native层MessageQueue的nativePollOnce(),它实际调用的是native层MessageQueue的pollOnce(),而native的pollOnce调用的是Native层的Looper的pollOnce:

static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj,
        jlong ptr, jint timeoutMillis) {
    ……
    nativeMessageQueue->pollOnce(env, obj, timeoutMillis);
}

void NativeMessageQueue::pollOnce(JNIEnv* env, jobject pollObj, int timeoutMillis) {
    ……
    mLooper->pollOnce(timeoutMillis);
    ……
}

在看Native层Looper的pollOnce方法之前,先看看Native层的Looper和Java层的Looper会不会有一些不一样吧。

Looper Native

和Java层Looper的使用一样,Native层Looper也需要prepare,也是一个通过线程局部变量存储的对象,一个线程只有一个。那么在Native层是怎么实现线程局部变量的呢?

Linux TSD(Thread-specific Data)池

Native层线程局部变量的思想和Java层很类似,Native层会维护一个全局的pthread_keys数组,用于存放线程局部变量的键。其中seq用于标记是否"in_use",destr则是一个函数指针,可用作析构函数,在线程退出时释放该键对应于线程中的线程局部变量。

static struct pthread_key_struct pthread_keys[PTHREAD_KEYS_MAX] ={{0,NULL}};

int pthread_key_create(pthread_key_t *key, void (*destr_function) (void*));

struct pthread_key_struct
{
  /* Sequence numbers.  Even numbers indicated vacant entries.  Note
     that zero is even.  We use uintptr_t to not require padding on
     32- and 64-bit machines.  On 64-bit machines it helps to avoid
     wrapping, too.  */
  uintptr_t seq;

  /* Destructor for the data.  */
  void (*destr) (void *);
};

pthread在创建线程时会维护一个指针数组,数组元素指向线程局部变量的数据块。整体解构如下图:

创建Looper

创建Looper时,会做以下事情:

  1. 通过eventfd创建mWakeEventFd用于线程间通信去唤醒Looper的,当需要唤醒Looper时,就往里面写1
  2. 创建用于监听epoll_event的mEpollFd,并初始化mEpollFd要监听的epoll_event类型
  3. 通过epoll_ctl将mWakeEventFd注册到mEpollFd中,当mWakeEventFd有事件可读则唤醒Looper
  4. 如果mRequests不为空的话,说明前面注册了有要监听的fd,则遍历mRequests中的Request,将它初始化为epoll_event并通过epoll_ctl注册到mEpollFd中,当有可读事件同样唤醒Looper
Looper::Looper(bool allowNonCallbacks) :
        mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),
        mPolling(false), mEpollFd(-1), mEpollRebuildRequired(false),
        mNextRequestSeq(0), mResponseIndex(0), mNextMessageUptime(LLONG_MAX) {
    mWakeEventFd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    ……
    rebuildEpollLocked();
}

void Looper::rebuildEpollLocked() {
    ……

    // Allocate the new epoll instance and register the wake pipe.
    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;
    int result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeEventFd, & 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, EPOLL_CTL_ADD, request.fd, & eventItem);
        ……
    }
}

pollOnce

对于Native层Looper的pollOnce,找它函数定义稍微有点隐秘,它在Looper.h中声明,inline到pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData)函数里了,它做了以下事情:

  1. 优先处理mResponses里的Response,即来自fd的事件
  2. 如果没有需处理的Response,再调用pollInner
inline int pollOnce(int timeoutMillis) {
    return pollOnce(timeoutMillis, NULL, NULL, NULL);
}

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) {
                ……
                return ident;
            }
        }

        if (result != 0) {
#if DEBUG_POLL_AND_WAKE
            ALOGD("%p ~ pollOnce - returning result %d", this, result);
#endif
            ……
            return result;
        }

        result = pollInner(timeoutMillis);
    }
}

pollInner这个函数比较长,它做了以下事情:

  1. 基于下一个Message调整获取Message的时间间隔timeoutMillis
  2. 清空mResponses
  3. 获取epoll事件,即将要处理的Message
  4. 更新mPolling,防止进入idle
  5. 执行合法性检查
  6. 如果epoll_event的fd为mWakeFd,说明是Looper的唤醒事件,则唤醒Looper
  7. 否则先将epoll_event封装为Request,更新epoll_event的事件类型,再封装为Response装入mResponses
  8. 循环取出mMessageEnvelopes队头的MessageEnvelope,并将MessageEnvelope中的Message交给对应的Native层的Handler处理
  9. 循环调用mResponses中所有Response的callback

至此对Android消息机制的学习就结束啦。

题外话

如果你觉得我的分享有帮助到你的话,请我吃个零食/喝杯咖啡呗~

One thought on “Android消息机制,从Java层到Native层剖析

发表评论

电子邮件地址不会被公开。 必填项已用*标注

跳至工具栏