Apache Kafka 是一款高吞吐、低延迟、可持久化、分区有序的分布式流式消息队列，相比于 RocketMQ，Kafka 在海量日志采集、实时数据流、大数据流式计算、海量埋点上报场景优势极大。

Java 生态 Kafka 教程泛滥，但高性能 C++ 网关、边缘采集服务、日志转发服务、游戏后端、大数据采集客户端 均需要使用官方 librdkafka 进行开发。

librdkafka 是 Kafka 官方推荐的 C/C++ SDK，具备极致吞吐、内存占用低、异步模型成熟、集群高可用等特性，是工业级 C++ 接入 Kafka 的唯一首选方案。

本文从零带你落地 Kafka C++ 全套开发能力，包含环境编译、基础生产消费、同步/异步发送、批量发送、分区有序、消息重试、失败回调、消费者位移管理、集群高可用、全局单例封装、生产场景选型与避坑，所有代码可直接编译上线。

一、Kafka C++ SDK 核心概述

1.1 什么是 librdkafka？

librdkafka 是 Apache Kafka 官方开源的高性能 C/C++ 客户端库，底层基于异步事件驱动模型，支持 Kafka 全部核心协议，跨平台、无虚拟机依赖，是目前性能最强、稳定性最高的 Kafka 客户端。

1.2 核心支持能力

• ✅ 同步发送、异步发送、批量聚合发送

• ✅ 分区策略、自定义分区投递、局部顺序消息

• ✅ 消费者自动分区负载均衡、重平衡机制

• ✅ 手动/自动 offset 位移提交

• ✅ 消息发送失败重试、异常回调捕获

• ✅ Kafka 集群多节点高可用自动容错

• ✅ 批量消费、消息头属性透传

• ✅ 全局单例客户端封装、连接池复用思想

1.3 Kafka vs RocketMQ 适用场景差异

• Kafka：极致高吞吐、海量日志、数据流、埋点、大数据实时计算，侧重流式处理

• RocketMQ：业务消息、订单、事务、死信、重试队列，侧重业务可靠性

1.4 C++ SDK 核心优势

• 无 JVM 依赖，无 GC 抖动，延迟极其稳定

• 事件驱动异步模型，单机支撑百万级 QPS

• 内存占用极低，适合长期驻留后台服务

• 完美适配 C++ 高性能服务技术栈

---

二、环境搭建 & 工程配置

2.1 系统依赖安装

sudo apt update
sudo apt install git cmake gcc g++ make libssl-dev zlib1g-dev -y

2.2 编译安装 librdkafka

# 拉取源码
git clone https://github.com/edenhill/librdkafka.git
cd librdkafka
mkdir build && cd build

# 编译安装
cmake -DCMAKE_BUILD_TYPE=Release ..
make -j$(nproc)
sudo make install
sudo ldconfig

2.3 项目目录结构

kafka_cpp_demo/
├── CMakeLists.txt
└── main.cpp

2.4 完整 CMakeLists.txt 配置

cmake_minimum_required(VERSION 3.16)
project(kafka_cpp_demo)
set(CMAKE_CXX_STANDARD 17)
set(CMAKE_CXX_STANDARD_REQUIRED ON)

# 头文件与库路径
include_directories(/usr/local/include)
link_directories(/usr/local/lib)

add_executable(kafka_demo main.cpp)

# 链接kafka依赖
target_link_libraries(kafka_demo
    rdkafka
    rdkafka++
    pthread
    ssl
    z
)

三、Kafka 基础核心概念

• Broker：Kafka 服务节点，集群多节点部署

• Topic：消息主题，业务数据隔离单元

• Partition：分区，Kafka 并发与有序性核心单元

• Offset：消息位移，消费者消费标记

• ConsumerGroup：消费者组，实现负载均衡

四、基础生产者实战（同步/异步/批量）

头文件统一引入，下文所有代码共用：

#include <iostream>
#include <string>
#include <vector>
#include <chrono>
#include <thread>
#include <librdkafka/rdkafkacpp.h>

using namespace std;

4.1 同步消息发送（业务可靠消息）

同步发送阻塞等待 broker 确认，适合业务可靠、不允许丢失的消息场景。

void KafkaSyncProduce()
{
    // 配置项
    string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);

    // Kafka集群地址，单机填写单个，集群用逗号分隔
    conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);

    // 创建生产者
    RdKafka::Producer* producer = RdKafka::Producer::create(conf, errstr);
    if (!producer)
    {
        cerr << "Producer 创建失败：" << errstr << endl;
        return;
    }

    string topic = "cpp_kafka_test_topic";
    string msg = "Kafka C++ 同步消息内容";

    // 同步发送
    RdKafka::ErrorCode resp = producer->produce(
        topic,
        RdKafka::PARTITION_UA,
        RdKafka::MSG_COPY,
        (void*)msg.c_str(), msg.size(),
        nullptr, 0,
        nullptr
    );

    if (resp != RdKafka::ERR_NO_ERROR)
    {
        cerr << "消息发送失败：" << RdKafka::err2str(resp) << endl;
    }
    else
    {
        cout << "同步消息发送成功" << endl;
    }

    // 刷新缓冲区，确保消息发出
    producer->flush(1000);

    delete producer;
    delete conf;
}

4.2 异步消息发送（高吞吐日志/埋点）

异步发送不阻塞业务线程，通过回调返回发送结果，适合海量埋点、日志上报、监控数据等高吞吐场景。

// 异步发送回调
class DeliveryReportCb : public RdKafka::DeliveryReportCb
{
public:
    void dr_cb(RdKafka::Message &msg) override
    {
        if (msg.err())
        {
            cerr << "异步消息发送失败：" << msg.errstr() << endl;
        }
        else
        {
            cout << "异步消息发送成功, partition:" << msg.partition() 
                 << ", offset:" << msg.offset() << endl;
        }
    }
};

void KafkaAsyncProduce()
{
    string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);

    conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);
    // 开启异步队列超时刷新
    conf->set("queue.buffering.max.ms", "5", errstr);

    static DeliveryReportCb dr_cb;
    conf->set("dr_cb", &dr_cb, errstr);

    RdKafka::Producer* producer = RdKafka::Producer::create(conf, errstr);
    if (!producer)
    {
        cerr << "Producer 创建失败：" << errstr << endl;
        return;
    }

    string topic = "cpp_kafka_test_topic";
    string msg = "Kafka C++ 异步高吞吐消息";

    // 异步投递
    producer->produce(
        topic,
        RdKafka::PARTITION_UA,
        RdKafka::MSG_COPY,
        (void*)msg.c_str(), msg.size(),
        nullptr, 0,
        nullptr
    );

    // 轮询触发回调
    producer->poll(0);
    this_thread::sleep_for(chrono::milliseconds(20));

    delete producer;
    delete conf;
}

4.3 批量消息发送（极致吞吐优化）

批量发送合并多条消息一次网络IO，大幅提升吞吐，是 Kafka 大数据采集场景标准用法。

void KafkaBatchProduce()
{
    string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);
    conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);

    static DeliveryReportCb dr_cb;
    conf->set("dr_cb", &dr_cb, errstr);

    // 批量聚合参数
    conf->set("batch.size", "16384", errstr);
    conf->set("linger.ms", "5", errstr);

    RdKafka::Producer* producer = RdKafka::Producer::create(conf, errstr);

    string topic = "cpp_kafka_test_topic";
    for (int i = 0; i < 20; i++)
    {
        string msg = "批量消息数据_" + to_string(i);
        producer->produce(
            topic, RdKafka::PARTITION_UA, RdKafka::MSG_COPY,
            (void*)msg.c_str(), msg.size(), nullptr, 0, nullptr
        );
        producer->poll(0);
    }

    producer->flush(1000);
    delete producer;
    delete conf;
    cout << "批量消息发送完成" << endl;
}

4.4 分区有序消息发送（业务顺序保障）

Kafka 同一分区内消息有序，通过自定义 key 哈希路由到固定分区，实现订单、流程类有序消息。

void KafkaOrderProduce()
{
    string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);
    conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);

    static DeliveryReportCb dr_cb;
    conf->set("dr_cb", &dr_cb, errstr);
    RdKafka::Producer* producer = RdKafka::Producer::create(conf, errstr);

    string topic = "cpp_kafka_order_topic";
    string order_key = "ORDER_10001";

    for (int i = 1; i <= 5; i++)
    {
        string msg = "订单流程步骤_" + to_string(i);
        producer->produce(
            topic,
            RdKafka::PARTITION_UA,
            RdKafka::MSG_COPY,
            (void*)msg.c_str(), msg.size(),
            order_key.c_str(), order_key.size(),
            nullptr
        );
        producer->poll(0);
    }

    producer->flush(1000);
    delete producer;
    delete conf;
    cout << "顺序消息发送完成" << endl;
}

五、基础消费者实战（自动位移提交）

void KafkaConsume()
{
    string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);
    RdKafka::Conf* tconf = RdKafka::Conf::create(RdKafka::CONF_TOPIC);

    // 集群地址
    conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);
    // 消费者组
    conf->set("group.id", "cpp_kafka_consumer_group", errstr);
    // 自动提交offset
    conf->set("enable.auto.commit", "true", errstr);
    conf->set("auto.commit.interval.ms", "1000", errstr);
    // 首次消费从头开始
    conf->set("auto.offset.reset", "earliest", errstr);

    RdKafka::KafkaConsumer* consumer = RdKafka::KafkaConsumer::create(conf, tconf, errstr);
    if (!consumer)
    {
        cerr << "消费者创建失败：" << errstr << endl;
        return;
    }

    // 订阅topic
    vector<string> topics = {"cpp_kafka_test_topic"};
    consumer->subscribe(topics);
    cout << "Kafka消费者启动成功，等待消息..." << endl;

    while (true)
    {
        RdKafka::Message* msg = consumer->consume(1000);
        if (msg->err() == RdKafka::ERR_NO_ERROR)
        {
            cout << "收到消息：" << (char*)msg->payload()
                 << " partition:" << msg->partition()
                 << " offset:" << msg->offset() << endl;
        }
        delete msg;
    }

    consumer->close();
    delete consumer;
    delete conf;
    delete tconf;
}

六、企业级高阶扩展功能（生产必备）

6.1 Kafka 集群高可用配置（多节点自动容错）

生产环境必须配置多 broker 节点，SDK 自动负载均衡、故障节点剔除、自动重连，彻底消除单点故障。

void KafkaClusterProduce()
{
    string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);
    // 多节点集群地址，逗号分隔
    string cluster_addr = "192.168.1.100:9092,192.168.1.101:9092,192.168.1.102:9092";
    conf->set("bootstrap.servers", cluster_addr, errstr);

    static DeliveryReportCb dr_cb;
    conf->set("dr_cb", &dr_cb, errstr);

    RdKafka::Producer* producer = RdKafka::Producer::create(conf, errstr);
    string msg = "Kafka集群高可用测试消息";

    producer->produce("cpp_kafka_cluster_topic",
        RdKafka::PARTITION_UA, RdKafka::MSG_COPY,
        (void*)msg.c_str(), msg.size(), nullptr, 0, nullptr);

    producer->flush(1000);
    delete producer;
    delete conf;
    cout << "集群消息发送成功" << endl;
}

6.2 手动 Offset 提交（精准消费保障）

自动提交存在消息丢失/重复消费风险，核心业务建议业务处理成功后手动提交位移。

void KafkaManualOffsetConsume()
{
    string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);
    RdKafka::Conf* tconf = RdKafka::Conf::create(RdKafka::CONF_TOPIC);

    conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);
    conf->set("group.id", "cpp_kafka_manual_group", errstr);
    // 关闭自动提交
    conf->set("enable.auto.commit", "false", errstr);
    conf->set("auto.offset.reset", "earliest", errstr);

    RdKafka::KafkaConsumer* consumer = RdKafka::KafkaConsumer::create(conf, tconf, errstr);
    consumer->subscribe({"cpp_kafka_test_topic"});

    cout << "手动位移消费者启动成功" << endl;

    while (true)
    {
        RdKafka::Message* msg = consumer->consume(1000);
        if (msg->err() == RdKafka::ERR_NO_ERROR)
        {
            // 模拟业务处理
            cout << "业务处理消息：" << (char*)msg->payload() << endl;

            // 业务成功，手动提交offset
            consumer->commitSync(*msg);
        }
        delete msg;
    }

    consumer->close();
    delete consumer;
    delete conf;
    delete tconf;
}

6.3 C++ 全局单例 Kafka 生产者封装（生产最优方案）

Kafka Producer 是重量级对象，禁止频繁创建销毁。封装线程安全懒汉单例，全局复用连接，避免句柄泄漏、连接风暴。

#include <mutex>
#include <memory>

class KafkaGlobalProducer
{
public:
    static KafkaGlobalProducer& Instance()
    {
        static KafkaGlobalProducer ins;
        return ins;
    }

    // 初始化全局生产者
    bool Init(const string& broker_list)
    {
        lock_guard<mutex> lock(mtx);
        if (m_producer) return true;

        string errstr;
        m_conf.reset(RdKafka::Conf::create(RdKafka::CONF_GLOBAL));
        m_conf->set("bootstrap.servers", broker_list, errstr);
        m_conf->set("queue.buffering.max.ms", "5", errstr);
        m_conf->set("batch.size", "16384", errstr);

        static DeliveryReportCb dr_cb;
        m_conf->set("dr_cb", &dr_cb, errstr);

        m_producer.reset(RdKafka::Producer::create(m_conf.get(), errstr));
        if (!m_producer)
        {
            cerr << "全局生产者初始化失败：" << errstr << endl;
            return false;
        }
        cout << "Kafka全局单例生产者初始化成功" << endl;
        return true;
    }

    // 通用发送接口
    bool SendMsg(const string& topic, const string& body, const string& key = "")
    {
        if (!m_producer) return false;

        RdKafka::ErrorCode ret = m_producer->produce(
            topic,
            RdKafka::PARTITION_UA,
            RdKafka::MSG_COPY,
            (void*)body.c_str(), body.size(),
            key.empty() ? nullptr : key.c_str(), key.size(),
            nullptr
        );
        m_producer->poll(0);
        return ret == RdKafka::ERR_NO_ERROR;
    }

    // 退出释放资源
    void Shutdown()
    {
        lock_guard<mutex> lock(mtx);
        if (m_producer)
        {
            m_producer->flush(1000);
            m_producer.reset();
        }
        if (m_conf) m_conf.reset();
    }

private:
    KafkaGlobalProducer() = default;
    ~KafkaGlobalProducer() { Shutdown(); }
    KafkaGlobalProducer(const KafkaGlobalProducer&) = delete;
    KafkaGlobalProducer& operator=(const KafkaGlobalProducer&) = delete;

    mutex mtx;
    unique_ptr<RdKafka::Conf> m_conf;
    unique_ptr<RdKafka::Producer> m_producer;
};

// 单例调用示例
void KafkaSingletonTest()
{
    // 服务启动初始化一次
    KafkaGlobalProducer::Instance().Init("127.0.0.1:9092");

    // 业务任意位置发送消息
    KafkaGlobalProducer::Instance().SendMsg("cpp_kafka_global_topic", "全局单例Kafka消息");
}

6.4 消息重试 + 死信队列(DLQ) 生产级完整实现

Kafka 原生无内置重试队列与死信队列机制，生产环境需手动实现重试策略+死信投递，解决业务瞬时失败重试、异常消息隔离、数据兜底修复问题。本节实现工业级方案：消费失败阶梯重试、最大重试次数拦截、异常消息转入死信Topic、死信独立消费、人工修复重发，完全对齐企业生产规范。

6.4.1 核心设计思路

• 自定义消息重试次数标记，限定最大重试次数（默认8次）；

• 业务异常不提交位移，利用Kafka天然重试机制重新消费；

• 消息达到最大重试次数后，不再重试，主动投递至专属死信Topic；

• 独立消费者监听死信队列，实现日志归档、异常告警、问题排查；

• 提供死信消息修复重发接口，人工修正后重新进入业务队列消费。

6.4.2 带重试机制的业务消费者

通过消息自定义Header透传重试次数，实现阶梯重试，规避无限重试阻塞问题。

// 全局重试配置常量
#define MAX_RETRY_TIMES 8          // 最大重试次数
#define Kafka_DLQ_TOPIC_SUFFIX "_DLQ" // 死信Topic后缀

// 获取消息Header中的重试次数
int GetMsgRetryTimes(RdKafka::Message& msg)
{
    // 首次消费无Header，默认重试0次
    if (!msg.headers())
        return 0;

    std::string retry_str;
    RdKafka::Headers* headers = msg.headers();
    // 提取自定义重试次数字段
    if (headers->get("retry_times", retry_str) == RdKafka::ERR_NO_ERROR)
    {
        return std::stoi(retry_str);
    }
    return 0;
}

// 给消息添加重试次数Header
void SetMsgRetryHeader(RdKafka::Producer* producer, RdKafka::Message& msg, int retry_times)
{
    RdKafka::Headers headers;
    headers.add("retry_times", std::to_string(retry_times));
    producer->headers_set(headers);
}

// 带重试、死信机制的核心消费者
void KafkaRetryConsumer()
{
    std::string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);
    RdKafka::Conf* tconf = RdKafka::Conf::create(RdKafka::CONF_TOPIC);

    // 集群配置
    conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);
    conf->set("group.id", "cpp_kafka_retry_consumer_group", errstr);
    conf->set("enable.auto.commit", "false", errstr);  // 关闭自动提交，手动管控
    conf->set("auto.offset.reset", "earliest", errstr);

    RdKafka::KafkaConsumer* consumer = RdKafka::KafkaConsumer::create(conf, tconf, errstr);
    if (!consumer)
    {
        std::cerr << "重试消费者创建失败：" << errstr << std::endl;
        return;
    }

    // 订阅业务Topic
    std::string biz_topic = "cpp_kafka_biz_topic";
    consumer->subscribe({biz_topic});
    std::cout << "带重试机制的业务消费者启动成功" << std::endl;

    // 初始化死信投递生产者
    RdKafka::Conf* dlq_conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);
    dlq_conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);
    RdKafka::Producer* dlq_producer = RdKafka::Producer::create(dlq_conf, errstr);

    while (true)
    {
        RdKafka::Message* msg = consumer->consume(1000);
        if (msg->err() != RdKafka::ERR_NO_ERROR)
        {
            delete msg;
            continue;
        }

        // 获取当前消息重试次数
        int cur_retry = GetMsgRetryTimes(*msg);
        std::cout << "消费消息，当前重试次数：" << cur_retry 
                 << " 消息内容：" << (char*)msg->payload() << std::endl;

        // 模拟业务异常（数据库故障、接口超时、数据异常等）
        bool biz_failed = true;
        if (!biz_failed)
        {
            // 业务处理成功，手动提交位移
            consumer->commitSync(*msg);
            std::cout << "消息处理成功，提交位移" << std::endl;
        }
        else
        {
            // 业务处理失败，判断是否达到最大重试次数
            if (cur_retry < MAX_RETRY_TIMES)
            {
                // 未达上限：重试投递，累加重试次数
                cur_retry++;
                SetMsgRetryHeader(dlq_producer, *msg, cur_retry);
                
                // 重新投递原Topic，实现重试
                dlq_producer->produce(
                    biz_topic,
                    msg->partition(),
                    RdKafka::MSG_COPY,
                    msg->payload(), msg->len(),
                    msg->key(), msg->key_len(),
                    nullptr
                );
                dlq_producer->poll(0);
                std::cout << "业务处理失败，第" << cur_retry << "次重试" << std::endl;
            }
            else
            {
                // 达到最大重试次数：转入死信队列
                std::string dlq_topic = biz_topic + Kafka_DLQ_TOPIC_SUFFIX;
                dlq_producer->produce(
                    dlq_topic,
                    RdKafka::PARTITION_UA,
                    RdKafka::MSG_COPY,
                    msg->payload(), msg->len(),
                    msg->key(), msg->key_len(),
                    nullptr
                );
                dlq_producer->poll(0);
                std::cerr << "重试次数耗尽，消息转入死信队列：" << dlq_topic << std::endl;
                
                // 放弃当前消息，提交位移，避免阻塞后续消费
                consumer->commitSync(*msg);
            }
        }
        delete msg;
    }

    consumer->close();
    delete consumer;
    delete dlq_producer;
    delete conf;
    delete tconf;
    delete dlq_conf;
}

6.4.3 死信队列独立消费者（告警+归档）

单独部署死信消费者，不影响主业务链路，实现死信消息监控、日志持久化、异常告警。

void KafkaDlqConsumer()
{
    std::string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);
    RdKafka::Conf* tconf = RdKafka::Conf::create(RdKafka::CONF_TOPIC);

    conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);
    conf->set("group.id", "cpp_kafka_dlq_consumer_group", errstr);
    conf->set("enable.auto.commit", "true", errstr);
    conf->set("auto.offset.reset", "earliest", errstr);

    RdKafka::KafkaConsumer* consumer = RdKafka::KafkaConsumer::create(conf, tconf, errstr);
    if (!consumer)
    {
        std::cerr << "死信消费者创建失败：" << errstr << std::endl;
        return;
    }

    // 订阅死信Topic
    std::string dlq_topic = "cpp_kafka_biz_topic" + Kafka_DLQ_TOPIC_SUFFIX;
    consumer->subscribe({dlq_topic});
    std::cout << "死信队列消费者启动成功，监听Topic：" << dlq_topic << std::endl;

    while (true)
    {
        RdKafka::Message* msg = consumer->consume(1000);
        if (msg->err() != RdKafka::ERR_NO_ERROR)
        {
            delete msg;
            continue;
        }

        // 死信消息业务处理
        std::cerr << "【KAFKA死信消息】" << std::endl;
        std::cerr << "消息内容：" << (char*)msg->payload() << std::endl;
        std::cerr << "消息分区：" << msg->partition() 
                 << " 消息位移：" << msg->offset() << std::endl;

        // 拓展业务：
        // 1. 死信消息入库归档，留存溯源数据
        // 2. 调用钉钉/企业微信接口发送异常告警
        // 3. 统计死信数量，监控服务异常率

        delete msg;
    }

    consumer->close();
    delete consumer;
    delete conf;
    delete tconf;
}

6.4.4 死信消息人工修复重发工具

针对格式错误、瞬时故障导致的死信消息，修复数据后重新投递至业务队列，恢复业务流转。

// 死信消息修复重发
bool KafkaDlqResend(const std::string& biz_topic, const std::string&amp& msg_body, const std::string&amp key = "")
{
    std::string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::CONF_GLOBAL);
    conf->set("bootstrap.servers", "127.0.0.1:9092", errstr);

    RdKafka::Producer* producer = RdKafka::Producer::create(conf, errstr);
    if (!producer)
    {
        std::cerr << "重发生产者创建失败：" << errstr << std::endl;
        return false;
    }

    // 修复后重新投递业务Topic，重置重试次数
    RdKafka::ErrorCode ret = producer->produce(
        biz_topic,
        RdKafka::PARTITION_UA,
        RdKafka::MSG_COPY,
        (void*)msg_body.c_str(), msg_body.size(),
        key.empty() ? nullptr : key.c_str(), key.size(),
        nullptr
    );

    producer->flush(1000);
    delete producer;
    delete conf;

    if (ret == RdKafka::ERR_NO_ERROR)
    {
        std::cout << "死信消息修复重发成功" << std::endl;
        return true;
    }
    else
    {
        std::cerr << "死信消息重发失败：" << RdKafka::err2str(ret) << std::endl;
        return false;
    }
}

// 重发测试示例
void DlqResendTest()
{
    // 修复异常消息内容
    std::string fix_msg = "修复后的正常业务消息";
    KafkaDlqResend("cpp_kafka_biz_topic", fix_msg, "biz_key_001");
}

6.4.5 重试&死信机制生产规范

• 重试次数配置：普通业务8次重试，核心金融业务可调整为12次，搭配阶梯重试间隔；

• 死信Topic规范：统一使用「业务Topic+_DLQ」命名，便于统一管理和监控；

• 禁止死信堆积：必须配置死信告警，每日巡检、定期清理归档；

• 严格手动提交：重试、死信场景必须关闭自动提交，杜绝消息丢失、重复消费；

• 消息溯源：死信消息必须留存原始Key、时间、分区、位移信息，方便问题排查。

七、完整 Main 函数测试入口

int main()
{
    // 基础发送
    KafkaSyncProduce();
    KafkaAsyncProduce();
    KafkaBatchProduce();
    KafkaOrderProduce();

    // 集群高可用测试
    KafkaClusterProduce();

    // 全局单例测试
    KafkaSingletonTest();

    // 基础消费者（单独运行）
    // KafkaConsume();
    // KafkaManualOffsetConsume();

    // 重试&死信队列测试（单独部署运行）
    // KafkaRetryConsumer();
    // KafkaDlqConsumer();
    // DlqResendTest();

    return 0;
}

八、业务场景精准选型

8.1 同步消息适用场景

• 核心业务数据、计费日志、关键事件上报

• 需要确保消息 100% 投递成功的场景

8.2 异步/批量消息适用场景

• 用户行为埋点、操作日志、访问流量上报

• 服务器监控指标、设备采集数据

• 大数据实时流式计算数据源

8.3 顺序消息适用场景

• 订单状态流转、业务流程审批

• 金融流水、账务变更记录

• IM 消息时序推送

8.4 手动 Offset 适用场景

• 数据落地数据库、文件写入等耗时业务

• 不允许重复消费、不允许丢失消息的核心链路

8.5 集群高可用+单例客户端场景

• 线上长期运行的 C++ 网关、采集服务、中间件服务

• 高并发、高可用、不允许单点故障的生产环境

---

九、生产最佳实践 & 避坑指南

• 禁止频繁创建 Producer：必须全局单例，避免端口句柄泄漏、连接风暴

• 高吞吐优先批量异步：开启 batch.size + linger.ms，极大提升吞吐

• 核心业务关闭自动提交：手动 commit 位移，杜绝消息丢失与重复消费

• 生产必须集群部署：多 broker 节点容错，杜绝单点故障

• 顺序消息必须指定 Key：保证同一业务 ID 路由同一分区

• 重试死信规范：业务异常统一走自定义重试机制，超次数强制转入死信，杜绝消息丢失与无限重试堆积

• 顺序消息必须指定 Key：保证同一业务 ID 路由同一分区

---

十、全文总结

1. librdkafka 是 C++ 接入 Kafka 的工业级高性能 SDK，适配所有高吞吐流式业务场景；

2. 日常开发可根据业务可靠性要求，选择同步、异步、批量、有序四种发送模式；

3. 生产环境必须配置多节点集群 + 全局单例生产者 + 手动位移提交，保障高可用与数据一致性；

4. Kafka 主打高吞吐流式数据处理，是日志采集、大数据实时计算、海量埋点场景的首选中间件。

互动提问：你在 C++ 接入 Kafka 时遇到过消息堆积、乱序、重复消费、连接泄漏等问题吗？欢迎评论区交流！