MessageQueue之分析native方法
2020-03-26 本文已影响0人
Lemon666
了解过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));
}
}
}
