线程消息通信与异步处理

http://blog.csdn.net/yujun411522/article/details/46444869
本文出自:【yujun411522的博客】

public MyLooperThread extends Thread{
     private Handler mHandler;
     public void run(){
     //在实例化handler之前一定要先调用Looper.prepare()方法
     //1 Looper.prepare()方法
     Looper.prepare();
     //2 实例化handler
     //mHandler = new Handler(){
               public void handleMessage(Message msg){
                         //处理msg操作
                    }

          };
     //3 Looper.loop()不断从消息队列中取出来信息
     Looper.loop();
     }     
}

public class Looper {
   // sThreadLocal.get() will return null unless you've called prepare().
    static final ThreadLocal<Looper> sThreadLocal = new ThreadLocal<Looper>();
    final MessageQueue mQueue;//内部维护一个消息队列
    final Thread mThread;//对应的线程
    volatile boolean mRun;//运行状态
    private static Looper mMainLooper = null;  // guarded by Looper.class，主线程的Looper

   public static void prepare() {
        if (sThreadLocal.get() != null) {//之前已经设置过Looper对象了，报异常，说明只能设置一次
            throw new RuntimeException("Only one Looper may be created per thread");
        }
        sThreadLocal.set(new Looper());//为sThreadLocal变量设置Looper，这个Looper必须有，且只能有一个，不能重复设置
    }
}

private Looper() {//私有方法，不允许外部访问它
        mQueue = new MessageQueue();//构造一个MessageQueue对象，Looper中维护了一个消息队列，这一点很重要
        mRun = true;//设置mRun为true
        mThread = Thread.currentThread();//mThread设置为当前线程
    }

 MessageQueue() {
        nativeInit();//这是一个本地方法
    }

class NativeMessageQueue {
public:
    NativeMessageQueue();
    ~NativeMessageQueue();

    inline sp<Looper> getLooper() { return mLooper; }//返回该Looper变量，这个Looper类是native中的Looper类，不是java层的

    void pollOnce(int timeoutMillis);//后面会分析，进行询问
    void wake();//后面会分析，向管道写入"W"

private:
    sp<Looper> mLooper;//维护一个Native层的Looper变量
};

static void android_os_MessageQueue_nativeInit(JNIEnv* env, jobject obj) {
     //obj为java层的MessageQueue对象
     //1 创建一个NativeMessageQueue对象,这个是native层的
    NativeMessageQueue* nativeMessageQueue = new NativeMessageQueue();
    if (! nativeMessageQueue) {
        jniThrowRuntimeException(env, "Unable to allocate native queue");
        return;
    }
    //2 将NativeMessageQueue和 java层的MessageQueue对象 关联起来，实际就是讲nativeMessagQueue对象保存到MessageQueue中的mPtr变量中
    android_os_MessageQueue_setNativeMessageQueue(env, obj, nativeMessageQueue);
}

NativeMessageQueue::NativeMessageQueue() {
    mLooper = Looper::getForThread();//调用getForThread，看返回值是否为null
    if (mLooper == NULL) {//如果返回值为null，则创建一个Looper并设置
        mLooper = new Looper(false);
        Looper::setForThread(mLooper);//
    }
}

Looper::Looper(bool allowNonCallbacks) :
        mAllowNonCallbacks(allowNonCallbacks), mSendingMessage(false),
        mResponseIndex(0), mNextMessageUptime(LLONG_MAX) {
    int wakeFds[2];
    //创建一个管道，读写端分别为wakeFds[0]和wakeFds[1] 
    int result = pipe(wakeFds); 

    mWakeReadPipeFd = wakeFds[0];//读端
    mWakeWritePipeFd = wakeFds[1];//写端

    result = fcntl(mWakeReadPipeFd, F_SETFL, O_NONBLOCK);//设置非阻塞方式
    result = fcntl(mWakeWritePipeFd, F_SETFL, O_NONBLOCK);//设置非阻塞方式 
 
    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 = mWakeReadPipeFd;
    result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mWakeReadPipeFd, & eventItem);//只监听mWakeReadPipeFd ,也就是读端
    }

static void android_os_MessageQueue_setNativeMessageQueue(JNIEnv* env, jobject messageQueueObj,
        NativeMessageQueue* nativeMessageQueue) {
    env->SetIntField(messageQueueObj, gMessageQueueClassInfo.mPtr,
             reinterpret_cast<jint>(nativeMessageQueue));
}

static struct {
    jfieldID mPtr;   // native object attached to the DVM MessageQueue
} gMessageQueueClassInfo;

int register_android_os_MessageQueue(JNIEnv* env) {
    int res = jniRegisterNativeMethods(env, "android/os/MessageQueue",
            gMessageQueueMethods, NELEM(gMessageQueueMethods));
   
    jclass clazz;
    FIND_CLASS(clazz, "android/os/MessageQueue");
    //mPtre指向Java层中MessageQueue的mPtr成员变量
    GET_FIELD_ID(gMessageQueueClassInfo.mPtr, clazz,"mPtr", "I");    
    return 0;
}

 public Handler() {
         ....
        //返回当前线程的Looper对象
        mLooper = Looper.myLooper();
        if (mLooper == null) {
            throw new RuntimeException(
                "Can't create handler inside thread that has not called Looper.prepare()");
        }
        mQueue = mLooper.mQueue;//将Looper中的mQueue赋值给Handler的成员变量mQueue
        mCallback = null;
    }

 public boolean sendMessageAtTime(Message msg, long uptimeMillis)//uptimeMillis 绝对时间，开机时间作为基准
    {
        boolean sent = false;
        MessageQueue queue = mQueue;//looper中创建的MessageQueue
        if (queue != null) {
            msg.target = this;//指明msg的处理者为this
            sent = queue.enqueueMessage(msg, uptimeMillis);//调用MessageQueue  的enqueueMessage 方法
        }
        else {
          .....
        }
        return sent;
    }

 final boolean enqueueMessage(Message msg, long when) {
        ....
        final boolean needWake;
        synchronized (this) {
            if (mQuiting) {
              return false;//handler所在线程正在退出
            } else if (msg.target == null) {
                mQuiting = true;
            }

            msg.when = when;//指明消息何时处理
            //mMessages 消息队列头部，p为当前消息
            Message p = mMessages;
             // 消息队列为空或者新消息需要立马处理或者新消息处理时间早于队首消息的处理时间
            if (p == null || when == 0 || when < p.when) {//如果是这些情况，需要将新消息插入到队首立马处理
                msg.next = p;
                mMessages = msg;
                needWake = mBlocked; // new head, might need to wake up
            } else {//否则找到合适的位置插入，按照时间的顺序
                Message prev = null;
                while (p != null && p.when <= when) {
                    prev = p;
                    p = p.next;
                }
                msg.next = prev.next;
                prev.next = msg;
                needWake = false; // still waiting on head, no need to wake up
            }
        }
        if (needWake) {//新加入消息队列，且处于block状态，需要唤醒
            nativeWake(mPtr);//本地方法，mPtr 就是NativeMessageQueue的native地址
        }
        return true;
    }

static void android_os_MessageQueue_nativeWake(JNIEnv* env, jobject obj, jint ptr) {
     //ptr 就是NativeMessageQueue的地址
    NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
    return nativeMessageQueue->wake();
}

void NativeMessageQueue::wake() {
    mLooper->wake();
}

void Looper::wake() {
 
   if (mPendingWakeCount++ == 0) {
        mPendingWakeTime = systemTime(SYSTEM_TIME_MONOTONIC);
    }
 
   ssize_t nWrite;
    do {
        nWrite = write(mWakeWritePipeFd, "W", 1);//向mWakeWritePipeFd 写入一个"W"
    } while (nWrite == -1 && errno == EINTR);

    if (nWrite != 1) {
        if (errno != EAGAIN) {
            LOGW("Could not write wake signal, errno=%d", errno);
        }
    }
}

void Looper::sendMessageAtTime(nsecs_t uptime, const sp<MessageHandler>& handler,
        const Message& message) { 

    size_t i = 0;
    { 
        AutoMutex _l(mLock);
         //native层消息队列的大小
        size_t messageCount = mMessageEnvelopes.size();
         //按照时间先后顺序找到它插入的位置 
        while (i < messageCount && uptime >= mMessageEnvelopes.itemAt(i).uptime) {
            i += 1;
        }
        //利用MessageEnvelope  ，将发送时间，messageHandler和消息封装在MessageEnvelope  中
        MessageEnvelope messageEnvelope(uptime, handler, message);
        //插入到native消息队列中
        mMessageEnvelopes.insertAt(messageEnvelope, i, 1);

        if (mSendingMessage) {
            return;
        }
    }  
     
    if (i == 0) {
        wake();//和java层调用nativeWake 方法效果一样，向管道写端写入一个"W"
    }
}

struct MessageEnvelope {
        MessageEnvelope() : uptime(0) { }

        MessageEnvelope(nsecs_t uptime, const sp<MessageHandler> handler,
                const Message& message) : uptime(uptime), handler(handler), message(message) {
        }

        nsecs_t uptime; //执行事件
        sp<MessageHandler> handler;//执行handler
        Message message;//消息
    };

  public static void loop() {
        Looper me = myLooper();
        if (me == null) {//可以看出，必须在设置当前线程的Looper之后才能执行loop方法
            throw new RuntimeException("No Looper; Looper.prepare() wasn't called on this thread.");
        }
        MessageQueue queue = me.mQueue;
       
        .....       
        while (true) {
            //1 从MessageQueue中取出消息
            Message msg = queue.next(); // might block
            if (msg != null) {
                if (msg.target == null) {
                 return;//没有消息时退出
                }
                .....
                //派发消息
                msg.target.dispatchMessage(msg);

                ....
               //回收消息
                msg.recycle();
            }
        }
    }

final Message next() {      
        int nextPollTimeoutMillis = 0;//下一次poll的时间,0表示马上进行
        for (;;) {
             ...
            //本地方法nativePollOnce
            nativePollOnce(mPtr, nextPollTimeoutMillis);

            synchronized (this) {//同步              
                final long now = SystemClock.uptimeMillis();
                final Message msg = mMessages;//队首消息
                if (msg != null) {
                    final long when = msg.when;//队首消息的执行事件 ，这里假如是4s,now为5s
                    if (now >= when) {//说明队首消息已经到了或者过了执行事件，所以立马执行
                        mBlocked = false;
                        mMessages = msg.next;//取出队首消息，返回
                        msg.next = null;
                        
                        msg.markInUse();//标记正在使用
                        return msg;
                    } else {
                        //否则，假如队首消息执行时间when为10s,now为5s，则过when-now= 5s之后再询问

                        nextPollTimeoutMillis = (int) Math.min(when - now, Integer.MAX_VALUE);
                    }
                } else {
                    //消息队列中没有消息
                    nextPollTimeoutMillis = -1;
                }                 
            }              
        }
    }

static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj,
        jint ptr, jint timeoutMillis) {
    //其中ptr就是NativeMessageQueue对象地址，timeOutMills就是经过多长时间询问
    NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
    nativeMessageQueue->pollOnce(timeoutMillis);
}

void NativeMessageQueue::pollOnce(int timeoutMillis) {
    mLooper->pollOnce(timeoutMillis);
}

 inline int pollOnce(int timeoutMillis) {
        return pollOnce(timeoutMillis, NULL, NULL, NULL);
    }

int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) {
     //timeoutMillis, NULL, NULL, NULL 参数
    int result = 0;
    for (;;) {
        while (mResponseIndex < mResponses.size()) {
            const Response& response = mResponses.itemAt(mResponseIndex++);
            ALooper_callbackFunc callback = response.request.callback;
            if (!callback) {
                int ident = response.request.ident;
                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 = NULL;
            if (outData != NULL) *outData = NULL;
            return result;
        }
        //执行pollInner方法
        result = pollInner(timeoutMillis);
    }
}

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;
        } 
    int result = ALOOPER_POLL_WAKE;
    mResponses.clear();
    mResponseIndex = 0;
 
    // Wait for wakeAndLock() waiters to run then set mPolling to true.
    mLock.lock();
    while (mWaiters != 0) {
        mResume.wait(mLock);
    }
    mPolling = true;
    mLock.unlock();

    size_t requestedCount = mRequestedFds.size();
    int eventCount = poll(mRequestedFds.editArray(), requestedCount, timeoutMillis);//监控mRequestedFds 上的时间 
 
    for (int i = 0; i < eventCount; i++) {
        int fd = eventItems[i].data.fd;
        uint32_t epollEvents = eventItems[i].events;
        if (fd == mWakeReadPipeFd) {
            if (epollEvents & EPOLLIN) {
                awoken();//调用awoken方法
            } else {
                  ....
            }
        } else {
          .....
        }
    }
Done: ;
 
    ...
    return result;
}

void Looper::awoken() {
    char buffer[16];
    ssize_t nRead;
    do {
         //读管道写端写入的"W"
        nRead = read(mWakeReadPipeFd, buffer, sizeof(buffer));
    } while ((nRead == -1 && errno == EINTR) || nRead == sizeof(buffer));
}