React Native 代码阅读（八）：MessageQueueThread 消息队列

2019-02-22

前言

React Native 底层是由消息队列机制来驱动的，消息队列是 React Native 中一个非常重要的组成部分。在本文中，我们开始对消息队列的梳理。

在开发 React Native 应用的时候，我们会遇到一种特殊卡顿，此时 UI 线程还是 60 帧，但是视觉上视图元素却卡住不动。出现这种情况的原因就是消息队列拥塞了。

保持消息队列畅通，是 React Native 流畅体验的关键。

消息队列所对应的类是 MessageQueueThread，它是 Java 侧的一个接口。它的实现类是 MessageQueueThreadImpl。

不仅如此，MessageQueueThread 还有位于 C++ 侧的映射类：ReactAndroid/src/main/jni/react/jni/JMessageQueueThread.h。

下面我们开始分别来看。

MessageQueueThread 接口

位于 ReactAndroid/src/main/java/com/facebook/react/bridge/queue/MessageQueueThread.java

它的作用是给线程添加 Looper，以供执行 Runnable。

提供的能力包括：

void runOnQueue(Runnable runnable)

  • 在线程的队列中运行 Runnable
  • Runnable 将会加入队尾
  • 可以重复添加，不会去重

callOnQueue(final Callable<T> callable)

  • 在线程队列中运行 Callable（java.util.concurrent.Callable）
  • Runnable 将会加入队尾
  • 可以重复添加，不会去重

boolean isOnThread()

  • 当前线程是否是 MessageQueueThread

void assertIsOnThread()

  • 当前线程是否是 MessageQueueThread
  • 如果不是抛出异常

void assertIsOnThread(String message)

  • 当前线程是否是 MessageQueueThread
  • 如果不是抛出异常

void quitSynchronous()

  • 退出 MessageQueueThread
  • 若从 MessageQueueThread 调用，这会是线程做的最后一件事
  • 若从别的线程调用这个 MessageQueueThread 的 quitSynchronous 方法，那个别的线程将会被阻塞，直到完成退出

MessageQueueThreadImpl 类

介绍

MessageQueueThreadImpl 是 MessageQueueThread 接口的实现类。

MessageQueueThread 有两类：

  • 运行在主线程的
  • 运行在后台线程的

我们来看看他是怎么实现 MessageQueueThread 接口定义的功能的。

Looper Handler

MessageQueueThreadImpl 的消息队列就是我们 Android 开发所熟知的 Looper Handler 机制：

private MessageQueueThreadImpl(
        String name,
        Looper looper,
        QueueThreadExceptionHandler exceptionHandler) {
    mName = name;
    mLooper = looper;
    mHandler = new MessageQueueThreadHandler(looper, exceptionHandler);
    mAssertionErrorMessage = "Expected to be called from the '" + getName() + "' thread!";
}

从因此，runOnQueue 跟 callOnQueue 也是没什么悬念，就是想 mHandler 里面扔 Runnable（Callable 则是先用 Runnable 封一下再扔）:

public void runOnQueue(Runnable runnable) {
    if (mIsFinished) {
        FLog.w(
            ReactConstants.TAG,
            "Tried to enqueue runnable on already finished thread: '" + getName() +
                "... dropping Runnable.");
    }
    mHandler.post(runnable);
}

isOnThread 判断方式也是我们非常熟悉的：

public boolean isOnThread() {
    return mLooper.getThread() == Thread.currentThread();
}

构造方式

MessageQueueThreadImpl 的构造方式是私有的，只能通过其对外暴露的 create 方法创建。

我们先来看构造方法的签名：

public static MessageQueueThreadImpl create(
    MessageQueueThreadSpec spec,
    QueueThreadExceptionHandler exceptionHandler) {

它接收两个参数：

  • MessageQueueThreadSpec 是一个配置类
  • QueueThreadExceptionHandler 用于处理 Handler 异常

MessageQueueThreadSpec

MessageQueueThreadImpl 在构造时通过 MessageQueueThreadSpec 指定构造参数。我们先来看下这个构造类。

构造参数包含三个：

  • ThreadType mThreadType：线程类型
  • String mName：名称
  • long mStackSize：栈大小，默认为 0，表示不指定大小

其中 ThreadType 是一个枚举：

protected static enum ThreadType {
    MAIN_UI,
    NEW_BACKGROUND,
}

它提供了3个静态方法，用于创建：

  • MessageQueueThreadSpec newUIBackgroundTreadSpec(String name)
  • MessageQueueThreadSpec newBackgroundThreadSpec(String name)
  • MessageQueueThreadSpec newBackgroundThreadSpec(String name, long stackSize)

具体构造

下面我们回到 MessageQueueThreadImpl 的 create，看具体构造方式。

首先在 create 方法中，根据 MessageQueueThreadSpec 指定的类型，执行对应的创建方法：

public static MessageQueueThreadImpl create(
        MessageQueueThreadSpec spec,
        QueueThreadExceptionHandler exceptionHandler) {
    switch (spec.getThreadType()) {
        case MAIN_UI:
            return createForMainThread(spec.getName(), exceptionHandler);
        case NEW_BACKGROUND:
            return startNewBackgroundThread(spec.getName(), spec.getStackSize(), exceptionHandler);
        default:
            throw new RuntimeException("Unknown thread type: " + spec.getThreadType());
    }
}

其中：

  • 我们注意到一点，栈大小只有 BackgroundThread 会用到

createForMainThread

createForMainThread 内部的主要工作是：

获取主线程 Looper：

Looper mainLooper = Looper.getMainLooper();

创建 MessageQueueThreadImpl 实例：

final MessageQueueThreadImpl mqt =
    new MessageQueueThreadImpl(name, mainLooper, exceptionHandler);

指定线程优先级：

Process.setThreadPriority(Process.THREAD_PRIORITY_DISPLAY);

startNewBackgroundThread

startNewBackgroundThread 的代码如下：

private static MessageQueueThreadImpl startNewBackgroundThread(
    final String name,
    long stackSize,
    QueueThreadExceptionHandler exceptionHandler) {
    final SimpleSettableFuture<Looper> looperFuture = new SimpleSettableFuture<>();
    Thread bgThread = new Thread(null,
        new Runnable() {
            @Override
            public void run() {
            Process.setThreadPriority(Process.THREAD_PRIORITY_DISPLAY);
            Looper.prepare();

            looperFuture.set(Looper.myLooper());
            Looper.loop();
            }
        }, "mqt_" + name, stackSize);
    bgThread.start();

    Looper myLooper = looperFuture.getOrThrow();
    return new MessageQueueThreadImpl(name, myLooper, exceptionHandler);
}

其中：

  • 创建了一个线程
  • 运行线程
  • 取出线程的 Looper，创建 MessageQueueThreadImpl 实例

ReactQueueConfiguration 接口

MessageQueueThreadImpl.create 方法是创建 createForMainThread 的入口。我们顺着入口向外，看看哪里调用它，来创建消息队列。

结果我们找到了 ReactQueueConfiguration 接口。这个接口是 React Native 中关于线程的设置类，它明确说明了在 React Native 中有三条消息队列线程：

  1. UI Queue Thread：标准 Android UI 主线程，不可设置
  2. Native Modules Queue Thread：native module 被调用的线程和 Looper
  3. JS Queue Thread：JS 执行所在的线程和 Looper

ReactQueueConfigurationImpl 类

介绍

ReactQueueConfigurationImpl 是 ReactQueueConfiguration 的实现类。

它包含三个成员：

private final MessageQueueThreadImpl mUIQueueThread;
private final MessageQueueThreadImpl mNativeModulesQueueThread;
private final MessageQueueThreadImpl mJSQueueThread;

再看 ReactQueueConfigurationImpl 的构造函数：

private ReactQueueConfigurationImpl(
        MessageQueueThreadImpl uiQueueThread,
        MessageQueueThreadImpl nativeModulesQueueThread,
        MessageQueueThreadImpl jsQueueThread) {
    mUIQueueThread = uiQueueThread;
    mNativeModulesQueueThread = nativeModulesQueueThread;
    mJSQueueThread = jsQueueThread;
}

构造函数也是私有的，通过对外暴露一个 create 方法来创建。

ReactQueueConfigurationImpl.create

create 方法的签名为：

public static ReactQueueConfigurationImpl create(
    ReactQueueConfigurationSpec spec,
    QueueThreadExceptionHandler exceptionHandler) {

首先，创建一个 Map，key 是 spec，value 是线程：

Map<MessageQueueThreadSpec, MessageQueueThreadImpl> specsToThreads = MapBuilder.newHashMap();

创建主线程队列：

MessageQueueThreadSpec uiThreadSpec = MessageQueueThreadSpec.mainThreadSpec();
MessageQueueThreadImpl uiThread =
    MessageQueueThreadImpl.create(uiThreadSpec, exceptionHandler);
specsToThreads.put(uiThreadSpec, uiThread);

创建 js 线程队列：

MessageQueueThreadImpl jsThread = specsToThreads.get(spec.getJSQueueThreadSpec());
if (jsThread == null) {
    jsThread = MessageQueueThreadImpl.create(spec.getJSQueueThreadSpec(), exceptionHandler);
}

创建 Native Module 线程队列：

MessageQueueThreadImpl nativeModulesThread = 
    specsToThreads.get(spec.getNativeModulesQueueThreadSpec());
if (nativeModulesThread == null) {
    nativeModulesThread =
        MessageQueueThreadImpl.create(spec.getNativeModulesQueueThreadSpec(), exceptionHandler);
}

最后创建实例并返回：

return new ReactQueueConfigurationImpl(
    uiThread,
    nativeModulesThread,
    jsThread);

CatalystInstanceImpl

介绍

ReactQueueConfigurationImpl 在哪里被使用呢？答案是 CatalystInstanceImpl。我们又来到了这个重要的类。

ReactQueueConfigurationImpl 是它的一个成员：

private final ReactQueueConfigurationImpl mReactQueueConfiguration;

使用

CatalystInstanceImpl 具体是如何使用 ReactQueueConfigurationImpl 的呢？

首先在 CatalystInstanceImpl 的构造方法中，创建 ReactQueueConfigurationImpl 实例：

mReactQueueConfiguration = ReactQueueConfigurationImpl.create(
    reactQueueConfigurationSpec,
    new NativeExceptionHandler());

我们现在知道，这句方法调用完成后，三条消息队列线程就都已经创建（主线程不用创建）好了。

之后 CatalystInstanceImpl 取出了 Native Thread，存入 mNativeModulesQueueThread 成员：

mNativeModulesQueueThread = mReactQueueConfiguration.getNativeModulesQueueThread();

之后，用 mReactQueueConfiguration 创建的线程来初始化 Bridge：

initializeBridge(
    new BridgeCallback(this),
    jsExecutor,
    mReactQueueConfiguration.getJSQueueThread(),
    mNativeModulesQueueThread,
    mNativeModuleRegistry.getJavaModules(this),
    mNativeModuleRegistry.getCxxModules());

其中，initializeBridge 是一个 C++ 层的方法，不在我们今天的研究范围，先略过。

在这里我们只要知道，ReactQueueConfigurationImpl 通过创建的线程，被成功地分发出去，在 React Native 的各处被使用起来即可。

ReactContext

在 React Native 中还有一个重要的类持有这三条消息队列线程，那就是 ReactContext。

我们在《React Native 中的各种 Context》中曾经说到过他。

我们再来看下它的 initializeWithInstance 方法如下：

public void initializeWithInstance(CatalystInstance catalystInstance) {
    if (catalystInstance == null) {
        throw new IllegalArgumentException("CatalystInstance cannot be null.");
    }
    if (mCatalystInstance != null) {
        throw new IllegalStateException("ReactContext has been already initialized");
    }

    mCatalystInstance = catalystInstance;

    ReactQueueConfiguration queueConfig = catalystInstance.getReactQueueConfiguration();
    mUiMessageQueueThread = queueConfig.getUIQueueThread();
    mNativeModulesMessageQueueThread = queueConfig.getNativeModulesQueueThread();
    mJSMessageQueueThread = queueConfig.getJSQueueThread();
}

其中：

  • 保存了 mCatalystInstance 对象
  • 保存了 mUiMessageQueueThread 和 mJSMessageQueueThread

initializeWithInstance 是在哪里创建的呢？这个请看前文，已经分析地很透彻了。

C++ 侧

介绍

MessageQueueThread 这个类的复杂之处在于，不仅 Java 侧会调用它，C++ 侧也会调用它。

在本节中我们分析 C++ 侧的消息队列线程。

MessageQueueThread 接口对应 C++ 类

Java 层的 MessageQueueThread 也有对应的 C++ 类，也叫 MessageQueueThread，位于 ReactCommon/cxxreact/MessageQueueThread.h。

class MessageQueueThread {
 public:
  virtual ~MessageQueueThread() {}
  virtual void runOnQueue(std::function<void()>&&) = 0;
  // runOnQueueSync and quitSynchronous are dangerous.  They should only be
  // used for initialization and cleanup.
  virtual void runOnQueueSync(std::function<void()>&&) = 0;
  // Once quitSynchronous() returns, no further work should run on the queue.
  virtual void quitSynchronous() = 0;
};

我们可以看到完全一致。

但是我们可以看出，这个类只是一个普通的 C++ 类，没有用到 fbjni 的那一套，怎么映射的呢？

映射类在 JavaMessageQueueThread，位于 ReactAndroid/src/main/jni/react/jni/JMessageQueueThread.h：

class JavaMessageQueueThread : public jni::JavaClass<JavaMessageQueueThread> {
public:
  static constexpr auto kJavaDescriptor = "Lcom/facebook/react/bridge/queue/MessageQueueThread;";
};

奇怪？怎么一个类拆两边写呢？没关系，还有一个 JMessageQueueThread 将两者厝合起来：

class JMessageQueueThread : public MessageQueueThread {
public:
  JMessageQueueThread(alias_ref<JavaMessageQueueThread::javaobject> jobj);

  /**
   * Enqueues the given function to run on this MessageQueueThread.
   */
  void runOnQueue(std::function<void()>&& runnable) override;

  /**
   * Synchronously executes the given function to run on this
   * MessageQueueThread, waiting until it completes.  Can be called from any
   * thread, but will block if not called on this MessageQueueThread.
   */
  void runOnQueueSync(std::function<void()>&& runnable) override;

  /**
   * Synchronously quits the current MessageQueueThread. Can be called from any thread, but will
   * block if not called on this MessageQueueThread.
   */
  void quitSynchronous() override;

  JavaMessageQueueThread::javaobject jobj() {
    return m_jobj.get();
  }

private:
  global_ref<JavaMessageQueueThread::javaobject> m_jobj;
};

看到这里我们明白了，原来使用了一个代理模式。

其中有一点就要注意，JMessageQueueThread 这个类是 C++ 层所使用的消息队列线程，我们只需要认准他就可以了。

ReactAndroid/src/main/jni/react/jni/JMessageQueueThread.cpp 中对各个方法的实现我们不再赘述。都是在 C++ 如何通过 fbjni 操纵 Java 对象，感兴趣的可以看看。

JMessageQueueThread 在哪里使用

C++ 层的 JMessageQueueThread 在哪里使用呢？在 C++ 层的 CatalystInstanceImpl 中。

首先，CatalystInstanceImpl 有一个成员，持有 Native Module Thread：

std::shared_ptr<JMessageQueueThread> moduleMessageQueue_;

在 CatalystInstanceImpl 的 initializeBridge 时，它会将 Native Module Thread 保存下来：

moduleMessageQueue_ = std::make_shared<JMessageQueueThread>(nativeModulesQueue);

同时在 CatalystInstanceImpl 的 initializeBridge 中，也会把 JS Thread 传给 Instance 类：

instance_->initializeBridge(
    folly::make_unique<JInstanceCallback>(
    callback,
    moduleMessageQueue_),
    jseh->getExecutorFactory(),
    folly::make_unique<JMessageQueueThread>(jsQueue),
    moduleRegistry_);

Instance 还记得是什么吗？（如果忘了参见前面的文章《C++ 侧的 Instance 类》），可以将它理解为 JavaScript 运行时。

Instance 中的消息队列

介绍

下面我们聚焦于 Instance，因为他是 JS 运行时，而我们刚刚梳理到把消息队列传入。在本节中，我们就来看运行时是如何与消息队列产生作用的。

Instance::initializeBridge

在 Instance::initializeBridge 中，把 jsQueue 封装成 NativeToJsBridge 保存下来：

void Instance::initializeBridge(
    std::unique_ptr<InstanceCallback> callback,
    std::shared_ptr<JSExecutorFactory> jsef,
    std::shared_ptr<MessageQueueThread> jsQueue,
    std::shared_ptr<ModuleRegistry> moduleRegistry) {
  callback_ = std::move(callback);
  moduleRegistry_ = std::move(moduleRegistry);

  jsQueue->runOnQueueSync([this, &jsef, jsQueue]() mutable {
    nativeToJsBridge_ = folly::make_unique<NativeToJsBridge>(
        jsef.get(), moduleRegistry_, jsQueue, callback_);

    std::lock_guard<std::mutex> lock(m_syncMutex);
    m_syncReady = true;
    m_syncCV.notify_all();
  });

  CHECK(nativeToJsBridge_);
}

其中：

  • nativeToJsBridge_ 这个成员暂时可以看做是消息队列线程了。
  • Instance 的很多功能：注册 Bundle、调用 JS 方法，都是通过 nativeToJsBridge_ 来实现的

Instance 与 nativeToJsBridge_ 的交互，这是 React Native 中很重要的一个机制，在本文中受限于篇幅，我们点到为止，在后面的文章中再详细介绍这一过程。

小结

在本文中，我们介绍了消息队列线程的：

  1. 实现原理
  2. 创建
  3. 在 React Native 中的位置
  4. 在 C++ 层的映射关系
  5. Instance 与消息队列的核心交互为主（引出问题）