如何使用librdkafka C++客户端添加自定义拦截器

librdkafka C++ 封装没有单独提供拦截器相关的API，你可以通过获取底层C语言实例指针，直接调用C版本的拦截器接口实现需求，具体操作步骤如下：

1. 引入必要头文件 #

除了C++客户端的常规头文件外，额外引入C版本拦截器相关头文件：

#include <librdkafka/rdkafkacpp.h>
#include <librdkafka/rd_kafka_interceptor.h> // C版本拦截器接口头文件

2. 实现符合C接口要求的拦截器回调函数 #

如果需要在回调中使用C层的自定义逻辑，可以通过ic_opaque参数传递C对象指针，在回调内部转换后调用：

// on_send拦截器回调，签名必须和C接口要求完全一致
rd_kafka_resp_err_t custom_on_send_interceptor(rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, void *ic_opaque) {
    // 将传入的自定义上下文转回你自己的C++类实例指针
    YourCustomLogicClass *logic_ptr = static_cast<YourCustomLogicClass*>(ic_opaque);
    
    // 此处编写自定义拦截逻辑，例如统计发送量、修改消息头、打印审计日志等
    std::cout << "拦截待发送消息，分区：" << rkmessage->partition << "，消息长度：" << rkmessage->len << std::endl;
    
    // 返回RD_KAFKA_RESP_ERR_NO_ERROR表示消息正常流转，返回其他错误码会终止当前消息发送
    return RD_KAFKA_RESP_ERR_NO_ERROR;
}

3. 初始化Producer后添加拦截器 #

C++的RdKafka::Producer类提供了公开的*c_ptr()*方法，可以获取底层对应的C语言rd_kafka_t*实例指针，直接传入C的拦截器接口即可：

int main() {
    // 原有C++ Producer初始化逻辑
    std::string errstr;
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);
    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
    if (!producer) {
        std::cerr << "Producer创建失败：" << errstr << std::endl;
        delete conf;
        return 1;
    }
    delete conf;

    // 自定义上下文对象，可将任意需要的参数传递到拦截器回调中
    YourCustomLogicClass custom_logic;

    // 调用C接口添加拦截器
    rd_kafka_resp_err_t add_err = rd_kafka_interceptor_add_on_send(
        producer->c_ptr(), // 传入C++实例底层的C指针
        "custom_send_interceptor", // 自定义拦截器名称
        custom_on_send_interceptor, // 你实现的回调函数
        &custom_logic // 传递给回调的自定义上下文
    );
    if (add_err != RD_KAFKA_RESP_ERR_NO_ERROR) {
        std::cerr << "拦截器添加失败：" << rd_kafka_err2str(add_err) << std::endl;
        delete producer;
        return 1;
    }

    // 后续正常使用producer发送消息即可，所有待发消息都会经过上述拦截器

    // 销毁逻辑
    delete producer;
    return 0;
}

注意事项 #

• 其他类型拦截器（如ack回调拦截器、消费拦截器等）的添加逻辑完全一致，只需替换对应的C拦截器添加函数、实现对应签名的回调函数即可。
• 拦截器回调是同步执行的，不要在回调中编写耗时逻辑，避免拖慢消息发送/消费性能。
• 传递给拦截器的ic_opaque指针的生命周期需要大于等于Producer实例的生命周期，避免出现野指针问题。

内容的提问来源于stack exchange，提问作者Jiewei Yuan