不能直接用 librdkafka 同步 API 做高吞吐生产，因其阻塞入队、无批量攒批、无背压反馈、错误需轮询捕获，易丢消息或延迟飙升；高吞吐必须用 rd_kafka_producev() + 后台线程 + 合理批处理 + 严格 delivery report 处理。
为什么不能直接用 librdkafka 的同步 API 做高吞吐生产
同步 rd_kafka_produce() 在内部会阻塞等待本地队列入队完成，且默认不启用后台线程轮询，消息堆积时容易卡住主线程；更关键的是它无法批量攒批、缺乏背压反馈、也不暴露 dr_cb（delivery report callback）外的错误上下文，一旦 broker 拒绝或网络抖动，错误只能靠轮询 rd_kafka_poll() 捕获——这在高吞吐场景下极易丢消息或延迟飙升。
真正支撑高吞吐的路径只有一条：用 rd_kafka_producev() + 启用后台线程 + 配置合理批处理 + 严格处理 delivery report。
- 必须设置
"queue.buffering.max.messages"≥ 100000（默认 100000，但常被误调小） - 必须启用
"enable.background.poll"="true"（librdkafka v1.8+），否则dr_cb可能永远不触发 -
"batch.num.messages"建议设为 1000–5000，太小增加 broker 请求压力，太大增加端到端延迟 - 不要在
dr_cb里做耗时操作（如写磁盘、锁竞争），否则阻塞 librdkafka 内部线程，导致生产者假死
如何正确初始化并配置 rd_kafka_t 实例
配置项顺序和类型敏感，尤其字符串值必须全小写，布尔值必须是 "true"/"false" 字符串，数值必须是字符串形式（如 "100000"），错一个就静默降级为默认值。
关键配置组合示例：
立即学习“C++免费学习笔记（深入）”；
rd_kafka_conf_t *conf = rd_kafka_conf_new();
rd_kafka_conf_set(conf, "bootstrap.servers", "kafka-broker:9092", errstr, sizeof(errstr));
rd_kafka_conf_set(conf, "message.send.max.retries", "10", errstr, sizeof(errstr));
rd_kafka_conf_set(conf, "retry.backoff.ms", "100", errstr, sizeof(errstr));
rd_kafka_conf_set(conf, "queue.buffering.max.messages", "200000", errstr, sizeof(errstr));
rd_kafka_conf_set(conf, "queue.buffering.max.kbytes", "2097152", errstr, sizeof(errstr)); // 2MB
rd_kafka_conf_set(conf, "batch.num.messages", "2000", errstr, sizeof(errstr));
rd_kafka_conf_set(conf, "linger.ms", "5", errstr, sizeof(errstr));
rd_kafka_conf_set(conf, "enable.idempotence", "true", errstr, sizeof(errstr)); // 必开，否则无法保证 at-least-once
rd_kafka_conf_set(conf, "enable.background.poll", "true", errstr, sizeof(errstr)); // librdkafka ≥1.8 必须
rd_kafka_conf_set_dr_msg_cb(conf, delivery_report_callback); // 必须设回调
rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
if (!rk) { /* handle error */ }
注意：rd_kafka_conf_new() 后不能再调用 rd_kafka_conf_set() 设置已存在的 key，否则返回失败；所有配置必须在 rd_kafka_new() 前完成。
怎么安全地异步发送并避免内存泄漏
librdkafka 不拷贝你传入的 payload 和 key，而是直接持有指针，因此你必须确保这些内存生命周期覆盖整个投递过程（从 rd_kafka_producev() 调用到 dr_cb 触发）。常见错误是栈变量或临时 std::string 的 data() 被释放后回调仍访问——必然 crash。
- 推荐方案：用
std::shared_ptr包裹 payload，在dr_cb中通过msg_opaque持有，确保释放时机可控 -
rd_kafka_producev()返回 -1 表示立即失败（如队列满），此时msg_opaque仍归你管理，需自行清理；返回 0 表示已入队，后续由dr_cb负责 - 永远检查
rd_kafka_producev()返回值，不要假设“只要没 crash 就成功” - 避免频繁 new/delete，可预分配 payload buffer 或用对象池管理
shared_ptr
典型发送片段：
auto payload_ptr = std::make_shared<std::string>("my message body");
rd_kafka_producev(rk,
RD_KAFKA_V_TOPIC(topic),
RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
RD_KAFKA_V_VALUE(payload_ptr->data(), payload_ptr->size()),
RD_KAFKA_V_OPAQUE(payload_ptr.get()), // 注意：不是 payload_ptr，避免循环引用
RD_KAFKA_V_END);
对应 dr_cb 中用 std::shared_ptr 构造新实例来延长生命周期：
void delivery_report_callback(rd_kafka_t *, const rd_kafka_message_t *rkmessage, void *) {
if (rkmessage->err) {
// 处理失败
}
auto *ptr = static_cast<std::string*>(rkmessage->opaque);
if (ptr) {
// 此处可安全访问 *ptr，但别 delete —— shared_ptr 会自动析构
}
}
怎么监控和应对真实生产环境中的背压
高吞吐下最隐蔽的问题不是发送失败，而是生产速度持续超过 broker 接收能力，导致本地队列不断膨胀，最终 OOM 或触发 queue.buffering.max.messages 拒绝新消息。librdkafka 提供了两个关键指标：
-
rd_kafka_outq_len(rk)：当前待投递消息总数（含 batch 中未 flush 的），应持续监控，> 50000 就要告警 -
rd_kafka_get_metadata()配合超时可检测 broker 连通性，但代价高，建议每 30s 一次
背压应对策略必须在应用层实现：
不能等 rd_kafka_producev() 返回 -1 才反应——那时队列已满；应该定期采样 rd_kafka_outq_len()，当连续 3 次 > 80% 阈值（如 160000），主动 throttle：比如把消息暂存环形缓冲区、降低采集频率、或向上游返回 429。librdkafka 本身不提供流控接口，这块完全由你控制。
另外，"queue.buffering.max.kbytes" 和 "queue.buffering.max.messages" 必须成比例调整，否则可能因某一项先触顶而误判背压。
