构建混合技术栈消息队列：C++数据平面与Kotlin控制平面的JNI通信实践

分布式系统

文章字数: 3.5k

阅读时长: 14 分

在处理需要微秒级延迟的交易或事件流场景时，纯JVM系的消息中间件，即便是那些经过深度优化的产品，有时也会触碰到性能天花板。垃圾回收（GC）的STW（Stop-The-World）暂停、JIT编译的预热时间、以及Java内存模型在极致低延迟场景下的间接性，都可能成为瓶颈。我们的团队就面临这样一个挑战：需要一个能保证P99.9延迟在50微秒以下的超低延迟消息通道，而现有的纯Kotlin方案在压测下只能勉强维持在毫秒级别。

最初的构想是彻底转向C++，但这意味着放弃Kotlin生态的巨大优势：成熟的协程框架、强大的构建工具链、以及团队已经积累的大量业务库。这在工程上是不可接受的。因此，一个混合架构方案浮出水面：将系统的核心——数据平面（Data Plane），即消息的接收、存储和分发——用C++实现，以压榨硬件的每一分性能。而系统的控制平面（Control Plane），包括Topic管理、消费者协调、监控和客户端SDK，则继续使用Kotlin，利用其高生产力和生态优势。

这种架构的关键粘合剂是Java Native Interface (JNI)。尽管JNI名声不佳，常与复杂、易错、性能开销大等标签挂钩，但如果设计得当，它能成为连接JVM世界和原生代码的强大桥梁。这里的核心设计原则是：最小化JNI的调用次数，但最大化单次调用的数据吞吐量。我们不会为每一条消息都进行一次JNI调用，而是通过JNI初始化和管理一个由C++完全控制的共享内存或本地Socket通信的后端。

graph TD
    subgraph Kotlin/JVM Space [Kotlin/JVM 空间]
        A[Kotlin Control Plane] --> B{JNI Bridge};
        C[Kotlin Producer/Consumer SDK] --> B;
    end

    subgraph Native Space [C++ 原生空间]
        D[C++ Data Plane Engine]
        E[Lock-Free Ring Buffer] --> D;
        F[io_uring Network IO] --> D;
    end

    B -- 1. Initialize Engine & Manage Topics --> D;
    B -- 2. Bulk Data Transfer --> D;

    style A fill:#D5E8D4,stroke:#82B366
    style C fill:#D5E8D4,stroke:#82B366
    style D fill:#DAE8FC,stroke:#6C8EBF

第一步：定义C++数据平面的核心接口

在深入JNI的细节之前，必须先有一个纯粹的、与Java世界无关的C++核心。这个核心必须是可独立测试、高性能的。我们选择了一个基于无锁环形缓冲区（Lock-Free Ring Buffer）的内存队列作为存储核心，因为它在高并发的单生产者单消费者或多生产者单消费者场景下表现极佳。

MessageBroker.h 这是C++引擎的头文件，定义了我们希望从Kotlin层调用的核心功能。

// MessageBroker.h
#ifndef HYBRID_MQ_MESSAGEBROKER_H
#define HYBRID_MQ_MESSAGEBROKER_H

#include <string>
#include <vector>
#include <cstdint>
#include <memory>
#include <functional>

// 定义消息的回调函数类型
// 使用 std::function 方便在 C++ 内部传递回调
using MessageCallback = std::function<void(const std::vector<uint8_t>&)>;

class MessageBroker {
public:
    // 获取单例实例，确保在 JNI 调用中我们操作的是同一个引擎
    static MessageBroker& getInstance();

    // 禁止拷贝和赋值
    MessageBroker(const MessageBroker&) = delete;
    MessageBroker& operator=(const MessageBroker&) = delete;

    /**
     * @brief 创建一个 Topic
     * @param topic_name Topic 的名称，必须唯一
     * @return 0 表示成功，-1 表示已存在
     */
    int create_topic(const std::string& topic_name);

    /**
     * @brief 向指定 Topic 发布消息
     * @param topic_name Topic 名称
     * @param data 消息内容
     * @param len 消息长度
     * @return 0 表示成功，-1 表示 Topic 不存在
     */
    int publish(const std::string& topic_name, const uint8_t* data, size_t len);

    /**
     * @brief 订阅一个 Topic
     * 在真实项目中，这里会返回一个订阅ID，并且回调会关联到特定消费者
     * 为了简化，我们这里只允许一个订阅者
     * @param topic_name Topic 名称
     * @param callback 收到消息时的回调函数
     * @return 0 表示成功，-1 表示 Topic 不存在或已被订阅
     */
    int subscribe(const std::string& topic_name, MessageCallback callback);
    
    /**
     * @brief 停止所有操作并清理资源
     * 在JNI的 Unload 中调用
     */
    void shutdown();

private:
    // 私有构造函数，用于单例模式
    MessageBroker();
    ~MessageBroker();

    // 内部实现，使用 PImpl 模式隐藏细节
    class BrokerImpl;
    std::unique_ptr<BrokerImpl> pimpl_;
};

#endif //HYBRID_MQ_MESSAGEBROKER_H

这里的PImpl模式（Pointer to Implementation）是一个重要的C++实践，它将实现细节（如具体的队列实现、锁机制等）与头文件分离，减少了编译依赖，使得我们可以在不改动头文件的情况下修改内部实现。

第二步：构建JNI桥接层

现在，我们需要一座连接Kotlin和C++的桥。首先在Kotlin中定义native方法。

HybridMqBridge.kt

// HybridMqBridge.kt
package com.example.hybridmq

import java.io.Closeable

/**
 * JNI 桥接对象，负责加载本地库并定义 native 接口
 * 这是一个单例，与C++层的MessageBroker单例对应
 */
object HybridMqBridge : Closeable {

    init {
        // 在真实项目中，会从 JAR 中解压库文件到临时目录
        // 为了演示，我们假设库文件在 java.library.path 中
        System.loadLibrary("hybridmq_jni")
    }

    /**
     * 初始化 C++ 引擎。
     * 必须是第一个调用的 native 方法。
     * @return 返回引擎实例的句柄（指针），后续操作都需要它
     */
    @JvmStatic
    external fun nativeInit(): Long

    /**
     * 关闭并清理 C++ 引擎资源
     * @param handle nativeInit 返回的句柄
     */
    @JvmStatic
    external fun nativeShutdown(handle: Long)
    
    /**
     * 创建 Topic
     * @param handle 引擎句柄
     * @param topicName Topic 名称
     * @return 0 表示成功，-1 表示失败
     */
    @JvmStatic
    external fun nativeCreateTopic(handle: Long, topicName: String): Int

    /**
     * 发布消息。
     * 使用 Direct ByteBuffer 来实现零拷贝数据传递，这是关键性能点。
     * @param handle 引擎句柄
     * @param topicName Topic 名称
     * @param messageBuffer 包含消息内容的 Direct ByteBuffer
     * @param offset buffer 中的起始位置
     * @param length 消息长度
     * @return 0 表示成功，-1 表示失败
     */
    @JvmStatic
    external fun nativePublish(handle: Long, topicName: String, messageBuffer: java.nio.ByteBuffer, offset: Int, length: Int): Int

    /**
     * 订阅 Topic。
     * @param handle 引擎句柄
     * @param topicName Topic 名称
     * @param callback 回调接口，当C++层有新消息时，会调用此接口的 onMessage 方法
     * @return 0 表示成功，-1 表示失败
     */
    @JvmStatic
    external fun nativeSubscribe(handle: Long, topicName: String, callback: MessageConsumer): Int
    
    override fun close() {
        // Kotlin的 `use` 语法会自动调用 close, 但这里我们没有全局句柄
        // 实际应用中，句柄管理需要更完善
        println("Bridge closed. Ensure nativeShutdown is called.")
    }
}

/**
 * 消息消费者的回调接口
 */
fun interface MessageConsumer {
    /**
     * C++ 引擎通过 JNI 回调此方法
     * @param message 消息内容的字节数组
     */
    fun onMessage(message: ByteArray)
}

这里的 nativePublish 方法使用了 java.nio.ByteBuffer，特别是Direct Buffer。这是一个性能优化的关键。当使用Direct ByteBuffer时，JVM会在堆外分配内存，C++代码可以直接访问这块内存的地址，避免了数据从JVM堆拷贝到原生内存的开销。

接下来是实现这些native方法的C++代码。

HybridMqJni.cpp

#include <jni.h>
#include <string>
#include <vector>
#include <android/log.h> // 只是为了方便演示，实际项目中用 spdlog 等
#include "MessageBroker.h"

// JNI 回调的辅助结构体，用于在C++中保存Java回调对象
struct JniCallback {
    JavaVM* jvm;
    jobject callback_obj;
    jmethodID on_message_method;
};

// 将 C++ 的 std::string 转换为 JNI 的 jstring
static std::string jstringToString(JNIEnv* env, jstring jstr) {
    if (!jstr) return "";
    const jclass stringClass = env->GetObjectClass(jstr);
    const jmethodID getBytes = env->GetMethodID(stringClass, "getBytes", "(Ljava/lang/String;)[B");
    const jbyteArray stringJbytes = (jbyteArray) env->CallObjectMethod(jstr, getBytes, env->NewStringUTF("UTF-8"));
    
    size_t length = (size_t) env->GetArrayLength(stringJbytes);
    jbyte* pBytes = env->GetByteArrayElements(stringJbytes, NULL);
    
    std::string ret = std::string((char *)pBytes, length);
    env->ReleaseByteArrayElements(stringJbytes, pBytes, JNI_ABORT);
    
    env->DeleteLocalRef(stringJbytes);
    env->DeleteLocalRef(stringClass);
    return ret;
}

// 宏定义，方便获取 broker 实例
#define GET_BROKER(handle) reinterpret_cast<MessageBroker*>(handle)

extern "C" {

JNIEXPORT jlong JNICALL
Java_com_example_hybridmq_HybridMqBridge_nativeInit(JNIEnv* env, jclass clazz) {
    // 返回 MessageBroker 单例的指针地址，转换为 long
    return reinterpret_cast<jlong>(&MessageBroker::getInstance());
}

JNIEXPORT void JNICALL
Java_com_example_hybridmq_HybridMqBridge_nativeShutdown(JNIEnv* env, jclass clazz, jlong handle) {
    GET_BROKER(handle)->shutdown();
}

JNIEXPORT jint JNICALL
Java_com_example_hybridmq_HybridMqBridge_nativeCreateTopic(JNIEnv* env, jclass clazz, jlong handle, jstring topic_name) {
    std::string topic = jstringToString(env, topic_name);
    return GET_BROKER(handle)->create_topic(topic);
}

JNIEXPORT jint JNICALL
Java_com_example_hybridmq_HybridMqBridge_nativePublish(JNIEnv* env, jclass clazz, jlong handle, jstring topic_name, jobject message_buffer, jint offset, jint length) {
    std::string topic = jstringToString(env, topic_name);
    // 从 Direct ByteBuffer 直接获取原生内存地址，零拷贝！
    auto* buffer = reinterpret_cast<uint8_t*>(env->GetDirectBufferAddress(message_buffer));
    if (buffer == nullptr) {
        // 不是 Direct ByteBuffer，需要处理错误
        return -1; 
    }
    return GET_BROKER(handle)->publish(topic, buffer + offset, length);
}

JNIEXPORT jint JNICALL
Java_com_example_hybridmq_HybridMqBridge_nativeSubscribe(JNIEnv* env, jclass clazz, jlong handle, jstring topic_name, jobject callback) {
    // --- 关键部分：处理从 C++ 到 Java 的回调 ---
    JavaVM* jvm;
    env->GetJavaVM(&jvm);
    
    // 创建一个全局引用，防止被JVM GC回收
    // 这是一个常见的JNI陷阱，局部引用在native方法返回后会失效
    jobject global_callback = env->NewGlobalRef(callback);
    jclass callback_class = env->GetObjectClass(global_callback);
    jmethodID on_message_method = env->GetMethodID(callback_class, "onMessage", "([B)V");

    if (on_message_method == nullptr) {
        // 方法未找到，清理并返回错误
        env->DeleteGlobalRef(global_callback);
        return -1;
    }

    auto jni_callback_helper = std::make_shared<JniCallback>({jvm, global_callback, on_message_method});

    // 创建一个C++ lambda，它将捕获JNI回调所需的所有信息
    MessageCallback cpp_callback = [jni_callback_helper](const std::vector<uint8_t>& msg_data) {
        JNIEnv* thread_env;
        // C++ 线程需要先附加到 JVM 才能执行 Java 代码
        jint attach_status = jni_callback_helper->jvm->AttachCurrentThread(reinterpret_cast<void**>(&thread_env), nullptr);
        if (attach_status != JNI_OK) {
            // 附加失败，无法回调
            return;
        }

        // 将 C++ 的 vector<uint8_t> 转换为 Java 的 byte[]
        jbyteArray j_byte_array = thread_env->NewByteArray(msg_data.size());
        thread_env->SetByteArrayRegion(j_byte_array, 0, msg_data.size(), reinterpret_cast<const jbyte*>(msg_data.data()));

        // 调用 Java 对象的 onMessage 方法
        thread_env->CallVoidMethod(jni_callback_helper->callback_obj, jni_callback_helper->on_message_method, j_byte_array);

        // 清理局部引用
        thread_env->DeleteLocalRef(j_byte_array);

        // 从 JVM 分离线程
        jni_callback_helper->jvm->DetachCurrentThread();
    };

    std::string topic = jstringToString(env, topic_name);
    int result = GET_BROKER(handle)->subscribe(topic, cpp_callback);

    if (result != 0) {
        // 如果订阅失败，必须清理已创建的全局引用，防止内存泄漏
        env->DeleteGlobalRef(global_callback);
    }
    
    return result;
}

} // extern "C"

这段JNI代码处理了几个关键点：

句柄传递：nativeInit返回C++对象的指针（强制转换为long），Kotlin层将其作为句柄保存。后续所有调用都将此句柄传回，C++层再将其转回指针，从而操作同一个对象实例。这是一个比单例更好的实践，因为它允许多个独立的Broker实例。
零拷贝发布：nativePublish直接从Direct ByteBuffer获取内存地址，避免了JVM堆和原生堆之间的数据拷贝。
C++回调Java：这是最复杂的部分。
- NewGlobalRef：当需要在C++中长期持有Java对象引用时（例如回调），必须使用全局引用。否则，当native方法返回时，局部引用会失效，导致后续回调时JVM崩溃。
- AttachCurrentThread / DetachCurrentThread：如果C++的回调发生在非JNI调用线程（例如一个专门的C++分发线程），该线程必须先附加到JVM才能安全地调用Java方法。回调结束后再分离。这是一个绝对不能忽视的规则。
- 数据转换：演示了如何将C++的std::vector转换为Java的byte[]。

第三步：集成为一个可运行的示例

现在，我们可以编写一个Kotlin main函数来测试整个流程。

Main.kt

// Main.kt
package com.example.hybridmq

import java.nio.ByteBuffer
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import kotlin.text.Charsets.UTF_8

fun main() {
    println("Initializing Hybrid MQ Engine...")
    val handle = HybridMqBridge.nativeInit()
    if (handle == 0L) {
        println("Failed to initialize native engine.")
        return
    }
    println("Native engine initialized with handle: $handle")

    val topic = "trade.events.stream"
    val creationResult = HybridMqBridge.nativeCreateTopic(handle, topic)
    println("Topic '$topic' creation result: $creationResult")

    val messageCount = 5
    val latch = CountDownLatch(messageCount)

    // 1. 设置订阅者
    val consumer = MessageConsumer { messageBytes ->
        val receivedMessage = String(messageBytes, UTF_8)
        println("[Consumer] Received message: '$receivedMessage'")
        latch.countDown()
    }
    HybridMqBridge.nativeSubscribe(handle, topic, consumer)
    println("Consumer subscribed to topic '$topic'.")

    // 2. 启动生产者
    Thread.sleep(100) // 等待订阅者就绪

    // 使用 Direct ByteBuffer
    val directBuffer = ByteBuffer.allocateDirect(1024)
    println("Producer starting to publish messages...")
    for (i in 1..messageCount) {
        val message = "Event aaaaaa-$i"
        directBuffer.clear()
        val messageBytes = message.toByteArray(UTF_8)
        directBuffer.put(messageBytes)
        directBuffer.flip() // 准备读取

        HybridMqBridge.nativePublish(
            handle = handle,
            topicName = topic,
            messageBuffer = directBuffer,
            offset = 0,
            length = messageBytes.size
        )
        println("[Producer] Published: '$message'")
        Thread.sleep(10)
    }

    // 3. 等待消费者接收所有消息
    val receivedAll = latch.await(5, TimeUnit.SECONDS)
    if (receivedAll) {
        println("All messages received successfully.")
    } else {
        println("Error: Timed out waiting for messages.")
    }

    // 4. 清理资源
    println("Shutting down native engine...")
    HybridMqBridge.nativeShutdown(handle)
    println("Shutdown complete.")
}

要运行这个项目，你需要配置CMake来编译C++代码为动态链接库（在Linux上是.so，macOS是.dylib，Windows是.dll），并配置Gradle来确保Kotlin编译时能找到这个库。这是一个典型的跨语言项目构建配置，虽然繁琐，但对于任何严肃的JNI项目都是必需的。

方案的局限性与未来展望

这个原型验证了混合架构的可行性，但它距离生产环境还有很长的路要走。

JNI的复杂性与脆弱性：JNI的API非常底层，手动管理内存、全局引用和线程附着很容易出错，且难以调试。任何一个小的失误都可能直接导致JVM崩溃，而不是抛出可捕获的异常。在真实项目中，我们会使用JNA（Java Native Access）或更现代的Project Panama（当它稳定后）来简化这种交互，尽管可能会牺牲一点性能。
单机限制：当前的实现是纯单机的。要构建一个真正的分布式消息队列，Kotlin控制平面需要与ZooKeeper或Etcd等协调服务集成，负责Topic元数据、消费者组偏移量和Broker成员关系的管理。C++数据平面之间也需要实现复制协议（如Raft）来保证高可用。
持久化缺失：当前队列是纯内存的。下一步是在C++层使用内存映射文件（mmap）来实现高性能的持久化日志存储，这是Kafka等现代消息队列的核心技术之一。
序列化与协议：目前消息体是原始字节。在分布式系统中，需要一个明确定义的网络协议和高效的序列化方案（如Protobuf或FlatBuffers），这部分逻辑可以放在C++层以获得最佳性能。

尽管存在这些挑战，但这种Kotlin+C++的混合架构模式提供了一条清晰的路径：它允许我们在系统的非性能关键部分（管理、协调、业务逻辑）利用Kotlin的高效开发能力和丰富的生态，同时在系统的性能核心（I/O、数据结构、计算密集任务）上，无妥协地使用C++来追求极致性能。这是一种务实的工程权衡，它承认没有一种语言是万能的，并将正确的技术用于正确的场景。

分布式与中间件 Kotlin C++ JNI 高性能计算

构建PB级可观测性数据后端 TiDB与ClickHouse的混合架构选型与实现

2023-10-27 后端架构

可观测性 TiDB ClickHouse 后端数据工程

利用 Pulumi Automation API 为移动端构建动态隔离的 Serverless NLP 推理架构

2023-10-27 架构设计

Pulumi Serverless NLP 架构模式移动开发