安卓基础组件Looper - 02 native层面的剖析

native使用

Android Native Looper 机制 - 掘金 (juejin.cn)

/system/core/libutils/include/utils/Looper.h
/system/core/libutils/Looper.cpp

使用总结

总结一下 Native 层 Looper 的使用：

// 初始化Looper对象
sp<Looper> mLooper = Looper::prepare(false /*allowNonCallbacks*/);// 可选：文件描述符
// - 添加要检测的文件描述符，
//   当对应事件发生时，调用回调对象中的回调函数
mLooper->addFd(fd, 0, ALOOPER_EVENT_INPUT, handleReceiveCallback, this);
// - 删除要检测的文件描述符
mLooper->removeFd(inputChannel->getFd());// 进入休眠状态, 等待超时或唤醒的回调
mLooper->pollAll(timeoutMillis);// 主动唤醒
mLooper->wake();// 发送消息 mLooper->sendMessage(handler, message);
// 删除消息 mLooper->removeMessages(handler, message);
void IrisService::runInThread(std::function<void()> handler, std::string tag, bool log) {mLooper->sendMessage(new LambdaMessage(handler, tag), Message());void ReportHandler::scheduleStreamingReport() {mHandlerLooper->removeMessages(this, WHAT_TAKE_REPORT);mHandlerLooper->sendMessage(this, Message(WHAT_TAKE_REPORT));

Android Native 层的 Looper 机制，关注的重点是：

• 如何实现休眠与唤醒
• 如何封装通知

创建Looper

构造函数创建(不推荐)

使用举例

可以使用裸指针，但建议仍然保证一个线程只有一个Looper的要求。

• 看一下 安卓系统服务surfaceflinger中的做法：

这里在进程的开头，在主线程里创造了一个Looper。所以没有其他子线程，满足 一个线程只有一个Looper

// frameworks/native/services/surfaceflinger/tests/vsync/vsync.cpp
int main(int /*argc*/, char** /*argv*/)
{sp<Looper> loop = new Looper(false);loop->addFd(myDisplayEvent.getFd(), 0, ALOOPER_EVENT_INPUT, receiver,&myDisplayEvent);

• 其他看起来无保护的场景

其他一些地方，在构造函数或init过程中，确确实实是new Looper了。但也不会多于一个，仍然保证了线程（主线程）只有一个Looper。

// frameworks/base/core/jni/android_os_MessageQueue.cpp
NativeMessageQueue::NativeMessageQueue() :mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {mLooper = Looper::getForThread();if (mLooper == NULL) {mLooper = new Looper(false);Looper::setForThread(mLooper);}
}

为什么没用Looper::Prepare Looper::xxThread可能出于性能的考虑？

sp<::android::Looper> mLooper;
mLooper = new Looper(false);

源代码

构造函数

# /system/core/libutils/include/utils/Looper.h
# /system/core/libutils/Looper.cpp

Looper::Looper(bool allowNonCallbacks): mAllowNonCallbacks(allowNonCallbacks),mSendingMessage(false),mPolling(false),mEpollRebuildRequired(false),mNextRequestSeq(WAKE_EVENT_FD_SEQ + 1),mResponseIndex(0),mNextMessageUptime(LLONG_MAX) {mWakeEventFd.reset(eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC));rebuildEpollLocked();
}

Looper::prepare

更推荐的做法：Looper::prepare

• 如果当前线程已经通过初始化过 Looper（使用构造函数且未使用setForThread指定除外），会直接返回对应Looper
• 设置Looper到当前线程，使用 RefBase::sp<Looper>智能指针管理Looper对象

// 初始化Looper对象
sp<Looper> mLooper = Looper::prepare(false /*allowNonCallbacks*/);

# /system/core/libutils/include/utils/Looper.h
# /system/core/libutils/Looper.cpp

Looper::Looper(bool allowNonCallbacks): mAllowNonCallbacks(allowNonCallbacks),mSendingMessage(false),mPolling(false),mEpollRebuildRequired(false),mNextRequestSeq(WAKE_EVENT_FD_SEQ + 1),mResponseIndex(0),mNextMessageUptime(LLONG_MAX) {mWakeEventFd.reset(eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC));rebuildEpollLocked();
}

sp<Looper> Looper::prepare(int opts) {bool allowNonCallbacks = opts & PREPARE_ALLOW_NON_CALLBACKS;sp<Looper> looper = Looper::getForThread();if (looper == nullptr) {looper = new Looper(allowNonCallbacks);Looper::setForThread(looper);}return looper;
}

void Looper::setForThread(const sp<Looper>& looper) {

sp<Looper> Looper::getForThread() {int result = pthread_once(& gTLSOnce, initTLSKey);LOG_ALWAYS_FATAL_IF(result != 0, "pthread_once failed");return (Looper*)pthread_getspecific(gTLSKey);
}

@TOC

获取Looper

获取当前线程已经绑定的Looper。

如果当前线程已经被 Looper::setForThread 方法指定了 Looper / 通过Looper::prepare初始化了Looper，那么可以通过 Looper::getForThread 方法获取

# /system/core/libutils/include/utils/Looper.h
# /system/core/libutils/Looper.cpp

static pthread_once_t gTLSOnce = PTHREAD_ONCE_INIT;
static pthread_key_t gTLSKey = 0;sp<Looper> Looper::getForThread() {int result = pthread_once(& gTLSOnce, initTLSKey);Looper* looper = (Looper*)pthread_getspecific(gTLSKey);return sp<Looper>::fromExisting(looper);
}void Looper::initTLSKey() {int error = pthread_key_create(&gTLSKey, threadDestructor);LOG_ALWAYS_FATAL_IF(error != 0, "Could not allocate TLS key: %s", strerror(error));
}void Looper::setForThread(const sp<Looper>& looper) {sp<Looper> old = getForThread(); // also has side-effect of initializing TLSif (looper != nullptr)looper->incStrong((void*)threadDestructor);pthread_setspecific(gTLSKey, looper.get());if (old != nullptr) {old->decStrong((void*)threadDestructor);}
}

可忽略

初始化Looper

Looper 的初始化，主要完成以下两个工作：

• 构建一个 epoll 池

• 维护epoll池

  • 构造一个 eventfd，放到 epoll 池
  • 把 mRequests 中保存的 fd 放到 epoll 池

  初始化时 mRequests 为空，这步可暂时忽略

(可选)文件描述符管理 fd

looper->addFd 添加需要关注的 fd （可选）

addfd 添加一个额外的 fd （eventfd 以外的）给 epoll 监听，主要完成了三项工作：

• 根据插入的 fd 回调对象等参数构建一个 Request 对象
• 把参数中的 fd 加入到 epoll 池中
• 把新构建的 Request 对象插入到 mRequests 中

// 可选：文件描述符
// - 添加要检测的文件描述符，
//   当对应事件发生时，调用回调对象中的回调函数
mLooper->addFd(fd, 0, ALOOPER_EVENT_INPUT, handleReceiveCallback, this);
// - 删除要检测的文件描述符
mLooper->removeFd(inputChannel->getFd());

主动休眠 pollAll

主动唤醒 wake

looper->pollAll 进入休眠状态等待回调、超时或唤醒

核心功能都聚集于 pollInner 函数中：

• 调用 epoll_wait 进入休眠状态

• 当 IO 事件到来时，被唤醒时

  • 读取事件，将发生的事件包装为 Response 对象，并保存到 mResponses 中

  • 处理所有的 Response，并回调 addfd 时传入的回调函数

  • 处理收到的 message，并调用回调函数

// 进入休眠状态, 等待超时或唤醒的回调
mLooper->pollAll(timeoutMillis);// 主动唤醒
mLooper->wake();

发送消息 sendMessage

其他线程是可以通过 Looper::sendMessage 给 Looper 所在工作线程发送消息

• looper 线程从休眠中唤醒，处理收到的 message，调用回调函数

• 主要完成了两项工作：

  • 将 message 截止时间 回调对象 包装为 MessageEnvelope 对象，并插入 mMessageEnvelopes。Looper::pollInner轮询时，就是从中取出Msg对象的。

    Vector<MessageEnvelope> mMessageEnvelopes; // guarded by mLock
    

  • 调用 wake 函数，给 eventfd 写数据，唤醒 epoll

void IrisService::runInSendingThread(std::function<void()> handler, std::string tag) {mSendingLooper->sendMessage(new LambdaMessage(handler, tag), Message());
}

# /system/core/libutils/include/utils/Looper.h
# /system/core/libutils/Looper.cpp

void Looper::sendMessage(const sp<MessageHandler>& handler, const Message& message) {nsecs_t now = systemTime(SYSTEM_TIME_MONOTONIC);sendMessageAtTime(now, handler, message);
}

void Looper::sendMessageAtTime(nsecs_t uptime, const sp<MessageHandler>& handler,const Message& message) {size_t i = 0;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);

轮询消息

native层面传入的message其实就是封装后的std::function——天然的可执行对象。

轮询实际上是自动完成的，会逐个执行这些可执行对象。

# /system/core/libutils/include/utils/Looper.h
# /system/core/libutils/Looper.cpp

Looper::pollAllLooper::pollOnceLooper::pollInner # 阻塞，轮询

int Looper::pollInner(int timeoutMillis) {// ...// 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 handlersp<MessageHandler> handler = messageEnvelope.handler;Message message = messageEnvelope.message;mMessageEnvelopes.removeAt(0);mSendingMessage = true;mLock.unlock();handler->handleMessage(message);} // release handlermLock.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;}}