librdkafka Producer发送Kafka消息报错，需补充哪些配置？

错误原因分析 #

你的代码存在几个关键问题导致该错误：

1. 未校验配置与对象创建结果：调用Conf::set、Producer::create、Topic::create后，既没检查操作是否成功，也没查看errstr中的具体错误信息，这是定位问题的核心遗漏点。
2. 缺少必要的生产者配置：仅设置bootstrap.servers不足以让生产者正常工作，至少需要指定消息确认策略（acks），否则librdkafka会因配置不完整拒绝执行发送操作。
3. 未处理消息发送后的轮询逻辑：生产者发送消息后需要调用poll()处理后台回调与事件，否则消息可能无法真正提交到Kafka，甚至触发配置类错误。

修复后的代码 #

#include <iostream>
#include <string>
#include <librdkafka/rdkafkacpp.h>

int main() {
    std::string brokers = "localhost:9092";
    std::string topic = "thirdtopic";
    std::string errstr;

    // 创建全局配置
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    if (!conf) {
        std::cerr << "创建全局配置失败" << std::endl;
        return 1;
    }

    // 设置Kafka集群地址，检查配置是否生效
    if (conf->set("bootstrap.servers", brokers, errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << "设置bootstrap.servers失败: " << errstr << std::endl;
        delete conf;
        return 1;
    }

    // 设置消息确认策略，可根据需求调整（1表示等待Leader节点确认）
    if (conf->set("acks", "1", errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << "设置acks失败: " << errstr << std::endl;
        delete conf;
        return 1;
    }

    // 创建生产者实例，检查是否成功
    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
    delete conf; // 配置已被生产者复用，可释放
    if (!producer) {
        std::cerr << "创建生产者失败: " << errstr << std::endl;
        return 1;
    }

    // 创建Topic配置
    RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
    if (!tconf) {
        std::cerr << "创建Topic配置失败" << std::endl;
        delete producer;
        return 1;
    }

    // 创建Topic实例，检查是否成功
    RdKafka::Topic *rd_topic = RdKafka::Topic::create(producer, topic, tconf, errstr);
    delete tconf;
    if (!rd_topic) {
        std::cerr << "创建Topic失败: " << errstr << std::endl;
        delete producer;
        return 1;
    }

    std::string message;
    std::cout << "输入消息或输入'exit'退出: ";
    std::getline(std::cin, message);

    if (message == "exit") {
        delete rd_topic;
        delete producer;
        return 0;
    }

    // 发送消息
    RdKafka::ErrorCode resp = producer->produce(
        rd_topic, 
        RdKafka::Topic::PARTITION_UA, 
        RdKafka::Producer::RK_MSG_COPY,
        const_cast<char*>(message.c_str()), 
        message.size(), 
        NULL, 
        NULL
    );

    if (resp != RdKafka::ErrorCode::ERR_NO_ERROR) {
        std::cerr << "发送消息失败: " << RdKafka::err2str(resp) << std::endl;
    } else {
        std::cout << "消息已进入发送队列: " << message << std::endl;
        // 轮询处理发送事件，确保消息提交到Kafka
        producer->poll(0);
    }

    // 等待所有未完成的消息发送完毕
    while (producer->outq_len() > 0) {
        std::cout << "等待剩余" << producer->outq_len() << "条消息发送完成" << std::endl;
        producer->poll(100);
    }

    delete rd_topic;
    delete producer;

    return 0;
}

发送消息时的必要配置补充 #

除bootstrap.servers外，以下是生产者核心配置项：

• acks：控制消息确认逻辑，可选值：
  • 0：不等待确认，性能最高但可能丢消息
  • 1：等待Leader节点确认，平衡性能与可靠性
  • all/-1：等待所有同步副本确认，可靠性最高
• retries：消息发送失败后的重试次数，默认值为2
• batch.size：批量发送的消息大小阈值，达到阈值后批量提交，提升传输效率
• linger.ms：批量发送的最大延迟时间，即使未达到batch.size也会触发发送
• client.id：生产者客户端标识ID，便于Kafka监控与日志排查
• compression.type：消息压缩类型，可选gzip、snappy、lz4等，减少网络传输量

额外注意事项 #

• 确保Kafka集群正常运行，且localhost:9092地址可正常访问
• 若目标Topic不存在，需确保Kafka开启auto.create.topics.enable=true（默认开启），或提前手动创建Topic
• 每次调用produce后，建议调用poll()处理后台事件，避免消息堆积

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