注该文用于个人学习记录和知识交流如有不足欢迎指点。一、Kafka 消费者核心配置说明1. 【必须设置】配置项无这些配置消费者无法运行配置项作用示例值bootstrap.servers指定 Kafka Broker 的地址列表多个用逗号分隔消费者通过此连接集群127.0.0.1:9092,192.168.1.100:9092group.id消费组 ID标识消费者所属的消费组是消费组模式分区自动分配的核心order_consumer_group如果使用assign手动分配分区非消费组模式group.id不是必须设置的配置项。消费组模式subscribe必须设置group.id。消费者会加入一个消费组由 Kafka 协调器自动管理分区分配group.id是消费组的唯一标识。手动分区模式assign可以不设置group.id。消费者绕过了消费组协调器直接绑定到指定的分区不需要加入任何消费组因此group.id不是运行的必要条件。注意如果在assign模式下仍然设置了group.id消费者会尝试加入消费组这可能导致与手动分配的分区产生冲突或意外的重平衡行为。代码示例无group.id的assign模式// 关键不设置 group.id conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); // conf-set(group.id, ..., errstr); // 这一行可以省略 // 手动创建 TopicPartition 列表并调用 assign std::vectorRdKafka::TopicPartition* partitions; partitions.push_back(RdKafka::TopicPartition::create(test_topic, 0)); partitions.push_back(RdKafka::TopicPartition::create(test_topic, 1)); RdKafka::KafkaConsumer* consumer RdKafka::KafkaConsumer::create(conf, errstr); consumer-assign(partitions); // 直接手动分配无需消费组模式是否必须设置group.id核心区别消费组模式(subscribe)✅是由 Kafka 协调器自动分配和管理分区。手动分区模式(assign)❌否消费者直接绑定到指定分区绕过消费组协调器。2. 【常用可选】配置项按需调整影响消费性能 / 可靠性配置项作用默认值推荐值enable.auto.commit是否自动提交 offset偏移量生产环境推荐关闭手动提交更可控truefalseauto.offset.reset当消费者无有效 offset 时的重置策略-latest从最新消息开始-earliest从最早消息开始latestearliestmax.poll.records单次consume/poll拉取的最大消息数批量拉取核心配置500100-1000session.timeout.ms消费组会话超时时间超时未心跳则被踢出消费组触发重平衡4500030000-60000heartbeat.interval.ms消费者向协调器发送心跳的间隔需小于 session.timeout.ms 的 1/3300010000fetch.min.bytes拉取消息的最小字节数Broker 攒够该字节数才返回提升吞吐量11024fetch.max.wait.msBroker 等待攒够fetch.min.bytes的最大时间超时必返回5001000max.partition.fetch.bytes单个分区单次拉取的最大字节数防止大消息撑爆内存104857652428805MBauto.commit.interval.ms自动提交的间隔仅enable.auto.committrue时生效5000100003. 核心总结必设配置仅 2 个bootstrap.servers连接集群、group.id消费组标识可选配置中enable.auto.commit手动提交、auto.offset.reset偏移量策略、max.poll.records批量拉取是生产环境最常调整的项直接影响消费可靠性和性能。二、全局配置跟主题配置rdkafka把配置拆分为全局CONF_GLOBAL和主题CONF_TOPIC本质是为了解决「不同粒度的配置需求」维度全局配置m_config主题配置m_topicConfig作用范围整个消费者实例所有主题、所有分区仅作用于具体的主题可全局默认 / 单独指定配置内容集群连接、消费组、回调函数、网络超时等全局属性位移重置策略、消息压缩、分区选择等主题级属性灵活性全局统一无法针对单个主题差异化可设置「全局默认值」也可给单个主题单独配置举个实际例子假设你的消费者要同时订阅topic_a和topic_bgroup.id消费组、bootstrap.servers集群地址必须全局统一用「全局配置」auto.offset.reset位移重置你想让topic_a消费历史消息earliesttopic_b只消费新消息latest这就需要「主题配置」来实现差异化。场景一 所有主题用相同的主题配置// 1. 创建全局配置m_config... 省略 auto* m_config RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL); ... // 2. 创建「默认主题配置」所有主题共用 RdKafka::Conf* default_topic_conf RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC); default_topic_conf-set(auto.offset.reset, latest, errorStr); // 所有主题都只消费新消息 m_config-set(default_topic_conf, default_topic_conf, errorStr); // 绑定到全局配置 // 3. 创建消费者 m_consumer RdKafka::KafkaConsumer::create(m_config, errorStr); // 4. 订阅两个主题topic_order 和 topic_pay 都会用上面的 latest 策略 std::vectorstd::string topics {topic_order, topic_pay}; m_consumer-subscribe(topics);当然不手动配置default_topic_conf也行karfka有默认的配置项内置默认值含义auto.offset.resetlatest无有效位移时只消费新消息enable.auto.committrue自动提交位移生产环境不推荐message.max.bytes1000000单条消息最大字节1MBrequest.timeout.ms30000主题请求超时时间30 秒但还是建议手写场景2 不同主题用不同的主题配置auto* m_config RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL); // 1. 全局配置 默认主题配置给 topic_pay 用 RdKafka::Conf* default_topic_conf RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC); default_topic_conf-set(auto.offset.reset, latest, errorStr); // topic_pay 用这个 m_config-set(default_topic_conf, default_topic_conf, errorStr); // 2. 为 topic_order 创建单独的主题配置 RdKafka::Conf* topic_order_conf RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC); topic_order_conf-set(auto.offset.reset, earliest, errorStr); // topic_order 用这个 // 3. 创建消费者 m_consumer RdKafka::KafkaConsumer::create(m_config, errorStr); // 4. 方式1用 Topic 对象订阅精准指定配置 RdKafka::Topic* topic_order RdKafka::Topic::create(m_consumer, topic_order, topic_order_conf, errorStr); RdKafka::Topic* topic_pay RdKafka::Topic::create(m_consumer, topic_pay, nullptr, errorStr); // 用默认配置 m_consumer-subscribe({topic_order, topic_pay}); // 传入 Topic 对象而非字符串 delete topic_order; delete topic_pay; // 方式2简化版先订阅所有主题再单独修改某个主题的配置部分版本支持 // m_consumer-subscribe({topic_order, topic_pay}); // m_consumer-topic_conf_set(topic_order, auto.offset.reset, earliest, errorStr);注意RdKafka::Topic* topic_pay RdKafka::Topic::create(m_consumer, topic_pay, nullptr, errorStr);中nullptr代表使用全局配置m_config中绑定的default_topic_confm_config-set(default_topic_conf, default_topic_conf, errorStr); // 2. 创建 topic_pay 时传入 nullptr表示用上面的默认配置 RdKafka::Topic* topic_pay RdKafka::Topic::create( m_consumer, // 消费者实例 topic_pay, // 主题名 nullptr, // 主题专属配置null 用默认 errorStr // 错误信息 );三、subscribe注意事项1.subscribe两种传参行为当你写m_consumer-subscribe({topic_order, topic_pay})时rdkafka内部会做这些事遍历每个字符串主题名为每个主题自动创建Topic对象且第三个参数主题配置传nullptr最终调用subscribe({自动创建的topic_order对象, 自动创建的topic_pay对象})。简单说传字符串 自动帮你创建「用默认配置的 Topic 对象」并订阅。传参方式底层行为使用的主题配置适用场景传Topic对象数组直接使用你创建的Topic对象每个Topic对象绑定的配置多主题差异化配置比如不同auto.offset.reset传字符串数组自动创建Topic对象配置 nullptr全局default_topic_conf或内置默认所有主题用相同配置极简写法关键注意点避坑无法实现差异化配置如果传字符串所有主题只能用同一个default_topic_conf没法给topic_order设earliest、topic_pay设latest错误处理差异传字符串时主题创建失败的错误会被封装到subscribe的返回值中传Topic对象时你可以提前检查每个Topic是否创建成功资源释放传字符串底层创建的Topic对象会被rdkafka自动释放无需你手动处理传Topic对象必须由你手动释放delete topic_order; delete topic_pay;否则会内存泄漏。2. subscribe 重复调用 覆盖订阅列表每次调用subscribe都会清空消费者当前的所有订阅关系然后绑定新的订阅列表。示例代码直观理解// 第一步订阅 topic_order std::vectorstd::string topics1 {topic_order}; m_consumer-subscribe(topics1); // 此时只消费 topic_order // 第二步重复调用 subscribe订阅 topic_pay std::vectorstd::string topics2 {topic_pay}; m_consumer-subscribe(topics2); // 此时只消费 topic_paytopic_order 被取消订阅 // 第三步再次重复调用订阅多个主题 std::vectorstd::string topics3 {topic_order, topic_pay}; m_consumer-subscribe(topics3); // 此时消费两个主题调用subscribe后消费者会立即触发「重平衡」Rebalance取消原订阅的所有主题 / 分区的分配重新向 Kafka 集群申请新订阅列表的分区如果传入空的字符串 / Topic 对象列表会取消消费者的所有订阅m_consumer-subscribe({}); // 停止消费所有主题进入空闲状态3.总结subscribe可重复调用行为是覆盖而非追加当前订阅列表每次调用会触发重平衡不要高频调用适用动态调整订阅的场景但需做好错误处理和位移提交空列表调用可取消所有订阅。