• Implementation in Java
  • ThreadLocal
  • MessageQueue
    • Asynchronous messages and message barriers
      • MessageQueue.postSyncBarrier
      • Consumption of message barriers
      • Role of asynchronous messages
  • Looper
  • Handler
  • MessageQueue.IdleHandler
• Implementation of Native
  • MessageQueue
  • Looper # creates an epoll event
  • epoll
  • nativePollOnce
  • nativeWake
  • Implementation of postDelay
• HandlerThread
• IntentService
• References and questions

outline

It is strongly recommended to read Gityuan's article

• Handler
• Looper
• ThreadLocal
• MessageQueue
• HandlerThread
• IntentService
• MessageQueue.IdleHandler
  • "Do you know the MessageQueue.IdleHandler of android?"

In short, when the message in the looper is temporarily processed, the interface will be called back. If false is returned, it will be removed. If true is returned, the callback will continue the next time the message is processed

/**
  * Callback interface for discovering when a thread is going to block
  * waiting for more messages.
*/
public static interface IdleHandler {
    /**
    * Called when the message queue has run out of messages and will now
    * wait for more.  Return true to keep your idle handler active, false
    * to have it removed.  This may be called if there are still messages
    * pending in the queue, but they are all scheduled to be dispatched
    * after the current time.
    */
    boolean queueIdle();
}

Implementation in Java

Four classes involved:

• Handler
• Looper
• ThreadLocal
• MessageQueue

ThreadLocal

Each thread has some variables related to itself. ThreadLocal is used to save these variables. All variables are stored through the internal static class Value. Although all threads access the same ThreadLocal, the variables saved by each thread are separate:

public void set(T value) {    
      Thread currentThread = Thread.currentThread();    
      Values values = values(currentThread);    
      if (values == null) {        
            values = initializeValues(currentThread);    }    
      values.put(this, value);
}

In the above set method, the values method returns the localValues member variable of the current thread:

/** 
  * Gets Values instance for this thread and variable type. 
  */
Values values(Thread current) {    
      return current.localValues;
}

So what are the local values inside the thread?

public class Thread implements Runnable {
    /** 
      * Normal thread local values. 
      */
    ThreadLocal.Values localValues;
    /*Omit several codes*/
}

As you can see, localValues in Thread are variables defined in ThreadLocal. If a null value is obtained in the values () method, the initializeValues method is executed. How is initializeValues implemented?

Values initializeValues(Thread current) {    
      return current.localValues = new Values();
}

Then put the value of value in the localValues of the current thread. In this way, although it seems that a ThreadLocal is accessed, the value obtained is different according to the thread.

Note: the internal implementation of ThreadLocal in different SDKs is different. For example, the implementation method in version 6.0 is not the above method, but the principle is the same

for instance

public class JsonTestMetaData {
    public String json_str;
}

public class MainActivity extends Activity {

    ThreadLocal<JsonTestMetaData> local = new ThreadLocal<>();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        Log.d(TAG, "onCreate");
       
        JsonTestMetaData data = new JsonTestMetaData();
        data.json_str = "main_thread";
        local.set(data);

        Log.e(TAG, local.get().json_str);

        new Thread(new Runnable() {
            @Override
            public void run() {
                JsonTestMetaData data = new JsonTestMetaData();
                data.json_str = "other_thread";
                local.set(data);
                Log.e(TAG, local.get().json_str);
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                if (local.get() != null) {
                    Log.e(TAG, local.get().json_str);
                } else {
                    Log.e(TAG, "local.get() is null");

                }
            }
        }).start();


        Log.e(TAG, local.get().json_str);

    }
}

Results obtained:

01-09 14:28:36.410 29303-29303/com.xx.app.javabcsxtest E/MainActivity: main_thread
01-09 14:28:36.412 29303-29303/com.xx.app.javabcsxtest E/MainActivity: main_thread
01-09 14:28:36.412 29303-29331/com.xx.app.javabcsxtest E/MainActivity: other_thread
01-09 14:28:36.413 29303-29332/com.xx.app.javabcsxtest E/MainActivity: local.get() is null

TheadLocal may cause a memory leak if the Thread does not exit.

MessageQueue

MessageQueue is a message queue that contains the member variable Message mMessages;, It can be understood as the head of the linked list. The storage form is not a queue, but a single linked list. The internal contains five native methods:

private native static long nativeInit();
private native static void nativeDestroy(long ptr);
private native static void nativePollOnce(long ptr, int timeoutMillis);
private native static void nativeWake(long ptr);
private native static boolean nativeIsIdling(long ptr);

The bottom layer is still completed through native code. The back part is expanded.

At the Java level, mainly

• Next method to obtain the next message;
• enqueueMessage inserts a message into a queue.

Asynchronous messages and message barriers

Such a piece of code is found in the scheduleTraversals method of ViewRootImpl

void scheduleTraversals() {
    if (!mTraversalScheduled) {
        mTraversalScheduled = true;
        mTraversalBarrier = mHandler.getLooper().getQueue().postSyncBarrier();
        mChoreographer.postCallback(
                Choreographer.CALLBACK_TRAVERSAL, mTraversalRunnable, null);
        if (!mUnbufferedInputDispatch) {
            scheduleConsumeBatchedInput();
        }
        notifyRendererOfFramePending();
        pokeDrawLockIfNeeded();
    }
}

It was found that a message barrier was added using MessageQueue

MessageQueue.postSyncBarrier

The code is relatively simple. It just inserts an Msg into the queue. when is SystemClock.uptimeMillis(), and the Msg has no target

    public int postSyncBarrier() {
        return postSyncBarrier(SystemClock.uptimeMillis());
    }

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

Consumption of message barriers

If the next method of MessageQueue encounters a message barrier, it will find the first asynchronous message after the linked list and give priority to asynchronous messages

Message next() {
    nativePollOnce(ptr, nextPollTimeoutMillis);
    synchronized (this) {
        Message prevMsg = null;
        Message msg = mMessages;

        
        if (msg != null && msg.target == null) {
            do {
                prevMsg = msg;
                msg = msg.next;
            } while (msg != null && !msg.isAsynchronous()); 
        }
        if (msg != null) {
            if (now < msg.when) {//Asynchronous messages are waiting before processing time (timeout)
                nextPollTimeoutMillis = (int) Math.min(msg.when - now, Integer.MAX_VALUE);
            } else {
                //When the asynchronous message reaches the processing time, it is removed from the linked list and returned.
                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 {
            nextPollTimeoutMillis = -1;
        }
        //...
    }
}

Role of asynchronous messages

You can take a look at the annotation of the setAsynchronous method of Message

In the process of invalid redrawing of View, a synchronization barrier will be introduced, which is actually to ensure that asynchronous messages are processed first, which is equivalent to jumping in the queue.?

Certain operations, such as view invalidation, may introduce synchronization barriers into the {@link Looper}'s message queue to prevent subsequent messages from being delivered until some condition is met. In the case of view invalidation, messages which are posted after a call to {@link android.view.View#invalidate} are suspended by means of a synchronization barrier until the next frame is ready to be drawn. The synchronization barrier ensures that the invalidation request is completely handled before resuming.

Asynchronous messages are exempt from synchronization barriers. They typically represent interrupts, input events, and other signals that must be handled independently even while other work has been suspended.

Looper

What does looper do? Looper's function is to create a MessageQueue for a thread, bind to this thread, and execute a message loop for this thread.

Looper contains references to MessageQueue and Thread

MessageQueue is created in the prepare method, and the loop starts in the loop method.

Looper and MessageQueue in Java layer have corresponding classes in C + +, namely looper (Native) and NativeMessageQueue classes

The thread has no looper by default. Unless you call the prepare method in the thread, the loop method can be executed for message processing.

What did prepare do?

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

You can see that the prepare() method can only be called once. Finally, a Looper will be created and saved in ThreadLocal. How was Looper created?

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

You can see that the construction method is private. A new MessageQueue is created, and mThread is the current thread. So how does Looper execute a message loop?

public static void loop() {    
        /*Omit several codes in the middle*/
        final Looper me = myLooper();    
        if (me == null) {        
              throw new RuntimeException("No Looper; Looper.prepare() wasn't called on this thread.");    
        }    
        final MessageQueue queue = me.mQueue;    
        for (;;) {        
              Message msg = queue.next();       
              msg.target.dispatchMessage(msg);     
              msg.recycleUnchecked();    
        }
}

You can see that through an infinite loop, you keep getting messages in the message queue and distributing them to the specified place.

The target of Message is actually the Handler

Therefore, in the thread you write, you can use the following:

class LooperThread extends Thread {     
      public Handler mHandler;      
      public void run() {          
            Looper.prepare();
            mHandler = new Handler() {
                  public void handleMessage(Message msg) {
                       // process incoming messages here
                  }
            };
            Looper.loop();
     }
}

However, the above method is too low and inconvenient in the code. It can be written as follows:

HandlerThread

HandlerThread thread  = new HandlerThread("new_handler_thread");
Handler handler = new Handler(thread.getLooper(), new Handler.Callback() {
            @Override
            public boolean handleMessage(Message msg) {
                return false;
            }
});

HandlerThread inherits from Thread and creates a Looper itself.

For HandlerThread, please refer to Android HandlerThread fully parsed

Most interactions with message loops are done through the Handler, as you can see in the Handler section. Remember to call Looper's quit method when you no longer execute the message loop.

Handler

Handler can send and process Message or Runnable associated with the MessageQueue. Each handler is associated with a separate thread, which is the thread where you created the handler, and the MessageQueue to be processed is also the MessageQueue of this thread.

See:

public Handler(Callback callback, boolean async) {    
    /*Omit several codes*/
    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 Handler(Looper looper, Callback callback, boolean async) {
    mLooper = looper;
    mQueue = looper.mQueue;
    mCallback = callback;
    mAsynchronous = async;
}

Handler has many constructors, which will eventually be called. You can see that the looper member in the handler is obtained through the looper's static method myloop. What does myloop do? You can see the contents of looper. In the above code, you get a looper associated with this thread. If the mloop member is null, an exception is thrown. You may ask, I just created a handler in the activity without calling the looper. Myloop () method. That's because when your application runs, Android has been created through Looper's static method prepareMainLooper. This method can only be executed once, otherwise an exception will be thrown. This looper is suitable for thread binding. Look at mQueue, which is obtained from mloop.

The sequence of calls is as follows:

• Is the callback of Message null? It is not the run method that calls the callback (in fact, the callback of Message here is a Runnable object, which is passed through the post method of the Handler)
• Is the callback of the Handler null? Not calling callback. The type of callback here is Handler.Callback()
• Finally, the method of Handler's handleMessage is called.

    /**
     * Handle system messages here.
     */
    public void dispatchMessage(Message msg) {
        if (msg.callback != null) {
            handleCallback(msg);
        } else {
            if (mCallback != null) {
                if (mCallback.handleMessage(msg)) {
                    return;
                }
            }
            handleMessage(msg);
        }
    }

MessageQueue.IdleHandler

• "Do you know the MessageQueue.IdleHandler of android?"

In short, when the message in the looper is temporarily processed, the interface will be called back. If false is returned, it will be removed. If true is returned, the callback will continue the next time the message is processed

Implementation of Native

Here you can refer to Gityuan - Android message mechanism 2-Handler(Native layer)

MessageQueue

MessageQueue.java

MessageQueue(boolean quitAllowed) {
    mQuitAllowed = quitAllowed;
    mPtr = nativeInit();  //mPtr records the information of the native message queue [2]
}

private native static long nativeInit();
private native static void nativeDestroy(long ptr);
private native void nativePollOnce(long ptr, int timeoutMillis);
private native static void nativeWake(long ptr);
private native static boolean nativeIsPolling(long ptr);
private native static void nativeSetFileDescriptorEvents(long ptr, int fd, int events);

You can see that the mPtr of the java layer is actually the address pointer of the native layer. About reinterpret_cast

static jlong android_os_MessageQueue_nativeInit(JNIEnv* env, jclass clazz) {
    //Initialize native message queue [3]
    NativeMessageQueue* nativeMessageQueue = new NativeMessageQueue();
    nativeMessageQueue->incStrong(env); //Increase reference count
    return reinterpret_cast<jlong>(nativeMessageQueue);
}

Unlike Java, there is a looper inside the native MessageQueue

NativeMessageQueue::NativeMessageQueue()
            : mPollEnv(NULL), mPollObj(NULL), mExceptionObj(NULL) {

    mLooper = Looper::getForThread(); //Gets the Looper object in TLS
    if (mLooper == NULL) {
        mLooper = new Looper(false); //Create Looper of native layer [4]
        Looper::setForThread(mLooper); //Save Looper of native layer to TLS
    }
}

• nativeInit() method:
  • The NativeMessageQueue object is created, its reference count is increased, and the NativeMessageQueue pointer mPtr is saved in the MessageQueue of the Java layer
  • A Native Looper object was created
  • Epoll calling epoll_create()/epoll_ctl() to complete the readable event listening of mWakeEventFd and mrrequests
• nativeDestroy() method
  • Call RefBase::decStrong() to reduce the reference count of the object
  • When the reference count is 0, the NativeMessageQueue object is deleted
• nativePollOnce() method
  • Call Looper::pollOnce() to complete, and stay in epoll when idle_ The wait () method is used to wait for an event to occur or time out
• nativeWake() method
  • Call Looper::wake() to complete and write characters to the pipeline mWakeEventfd;

Looper # creates an epoll event

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); //fd for constructing wake-up events
    AutoMutex _l(mLock);
    rebuildEpollLocked();  //Rebuild Epoll event [5]
}

eventfd

void Looper::rebuildEpollLocked() {
    if (mEpollFd >= 0) {
        close(mEpollFd); //Close the old epoll instance
    }
    mEpollFd = epoll_create(EPOLL_SIZE_HINT); //Create a new epoll instance and register the wake pipe
    struct epoll_event eventItem;
    memset(& eventItem, 0, sizeof(epoll_event)); //Set the unused data area to 0
    eventItem.events = EPOLLIN; //Readable event
    eventItem.data.fd = mWakeEventFd;
    //Add wake-up event (mWakeEventFd) to epoll instance (mEpollFd)
    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);
        //Add the events of the request queue to the epoll instance respectively
        int epollResult = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, request.fd, & eventItem);
    }
}

epoll

Gityuan - select/poll/epoll comparative analysis

The number of monitored descriptors is unlimited. The upper limit of FD supported is the maximum number of files that can be opened. The specific number can be viewed in cat / proc / sys / FS / file max. generally speaking, this number has a great relationship with the system memory. For 3G mobile phones, this value is 200000-300000.

IO performance does not decrease as the number of monitors fd increases. epoll is different from the method of select and poll polling, but is implemented through the callback function defined by each fd. Only the ready fd will execute the callback function.

nativePollOnce

Message next() {
    final long ptr = mPtr;
    if (ptr == 0) {
        return null;
    }

    for (;;) {
        ...
        nativePollOnce(ptr, nextPollTimeoutMillis); //Blocking operation [2]
        ...
    }

android_os_MessageQueue.cpp

static void android_os_MessageQueue_nativePollOnce(JNIEnv* env, jobject obj, jlong ptr, jint timeoutMillis) {
    //Convert the mPtr passed down from the Java layer to nativeMessageQueue
    NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
    nativeMessageQueue->pollOnce(env, obj, timeoutMillis); [3]
}

android_os_MessageQueue.cpp

void NativeMessageQueue::pollOnce(JNIEnv* env, jobject pollObj, int timeoutMillis) {
    mPollEnv = env;
    mPollObj = pollObj;
    mLooper->pollOnce(timeoutMillis); [4]
    mPollObj = NULL;
    mPollEnv = NULL;
    if (mExceptionObj) {
        env->Throw(mExceptionObj);
        env->DeleteLocalRef(mExceptionObj);
        mExceptionObj = NULL;
    }
}

Looper.h

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

Looper.cpp

int Looper::pollOnce(int timeoutMillis, int* outFd, int* outEvents, void** outData) {
    int result = 0;
    for (;;) {
        // Handle the Response event without the Callback method first
        while (mResponseIndex < mResponses.size()) {
            const Response& response = mResponses.itemAt(mResponseIndex++);
            int ident = response.request.ident;
            if (ident >= 0) { //If ident is greater than 0, there is no callback because POLL_CALLBACK = -2,
                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;
        }
        // Reprocess internal polling
        result = pollInner(timeoutMillis); [6]
    }
}

nativeWake

Call Looper::wake() to complete and write characters to the pipeline mWakeEventfd; (wake up event of epoll)

boolean enqueueMessage(Message msg, long when) {
    ... //Insert messages into MessageQueue in chronological order
    if (needWake) {
            nativeWake(mPtr); [2]
        }
}

static void android_os_MessageQueue_nativeWake(JNIEnv* env, jclass clazz, jlong ptr) {
    NativeMessageQueue* nativeMessageQueue = reinterpret_cast<NativeMessageQueue*>(ptr);
    nativeMessageQueue->wake(); [3]
}

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

void Looper::wake() {
    uint64_t inc = 1;
    // Write character 1 to pipeline mWakeEventFd
    ssize_t nWrite = TEMP_FAILURE_RETRY(write(mWakeEventFd, &inc, sizeof(uint64_t)));
    if (nWrite != sizeof(uint64_t)) {
        if (errno != EAGAIN) {
            ALOGW("Could not write wake signal, errno=%d", errno);
        }
    }
}

Implementation of postDelay

• 1. Call sendMessageAtTime to calculate the current time and delay time, and call enqueuemessage of MessageQueue to join the queue
• 2. enqueueMessage of messagequeue will traverse the queue and calculate the position inserted into the queue according to when of Message. If the next loop is blocked, it will call native method to wake up NativeMessageQueue
• 3.nativePollOnce will pass the time to the Looper.cpp method, which is actually epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, timeoutMillis); That is, the blocking time of epoll.

public final boolean postDelayed(Runnable r, long delayMillis)
{
    return sendMessageDelayed(getPostMessage(r), delayMillis);
}

public final boolean sendMessageDelayed(Message msg, long delayMillis)
{
    if (delayMillis < 0) {
        delayMillis = 0;
    }
    return sendMessageAtTime(msg, SystemClock.uptimeMillis() + delayMillis);
}
 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);
}   
private boolean enqueueMessage(MessageQueue queue, Message msg, long uptimeMillis) {
    msg.target = this;
    if (mAsynchronous) {
        msg.setAsynchronous(true);
    }
    return queue.enqueueMessage(msg, uptimeMillis);
} 

 boolean enqueueMessage(Message msg, long when) {
        if (msg.target == null) {
            throw new IllegalArgumentException("Message must have a target.");
        }
        if (msg.isInUse()) {
            throw new IllegalStateException(msg + " This message is already in use.");
        }

        synchronized (this) {
            if (mQuitting) {
                IllegalStateException e = new IllegalStateException(
                        msg.target + " sending message to a Handler on a dead thread");
                Log.w(TAG, e.getMessage(), e);
                msg.recycle();
                return false;
            }

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

HandlerThread

Inheriting from Thread, loop. Prepare() will be called when run ning; There are two important ways

public Looper getLooper()

public Handler getThreadHandler()

After using, you need to call quit()

IntentService

onCreate() will create the HandlerThread and start. The onHandleIntent method of servicehandler override of the Looper that created the HandlerThread just now is executed in the asynchronous thread

Relevant video recommendations:

[Android handle interview] how to create a Message in the Handler?
[Android handle interview] how does MessageQueue keep the thread safety of messages added by each thread?
Android (Android) development zero foundation tutorial from introduction to mastery: Studio installation / UI/Fragment / four components / popular framework / project release and management / project practice

This article is transferred from https://juejin.cn/post/6984960343130767373 , in case of infringement, please contact to delete.