Win32 C++集成librdkafka实战：从编译到生产消费完整指南

1. 项目概述与核心价值

最近在做一个Windows平台上的数据采集项目，需要将海量的设备日志实时推送到后端处理集群。消息队列选型上，团队毫不犹豫地定了Kafka，毕竟吞吐量和可靠性摆在那里。但客户端这块就有点头疼了：采集程序是用C++写的，跑在Win32环境（也就是我们常说的Windows桌面或服务器平台，非UWP那种）。搜了一圈，C++连接Kafka的主流库就是librdkafka，但网上的资料要么是Linux下的，要么就是语焉不详的代码片段，真正能在Windows上跑通、并且把生产消费流程讲清楚的实战内容太少了。踩了无数坑之后，我决定把从环境搭建、库编译、到生产消费核心代码编写的完整过程记录下来。如果你也在Win32下用C++折腾Kafka，这篇内容或许能帮你省下大半天甚至更久的摸索时间。

简单说，这个实战的目标就是：在Visual Studio的Win32项目里，集成librdkafka库，编写出稳定、高效的Kafka生产者和消费者程序。它解决的是Windows传统C++应用与现代大数据管道（Kafka）之间的桥接问题。无论你是做客户端数据上报、传统桌面软件的数据总线改造，还是嵌入式网关（跑Windows IoT）的数据转发，这套方案都直接适用。接下来，我会假设你熟悉C++基础，对Kafka的基本概念（Topic, Partition, Producer, Consumer）有所了解，然后我们一步步从零开始。

2. 环境准备与librdkafka编译

在Windows上玩C++开源库，第一道坎往往是编译。librdkafka官方并没有提供预编译的Windows二进制包，所以我们必须自己动手。别怕，过程虽然繁琐，但一步步来并不难。

2.1 工具链选择与准备

首先明确工具链。在Win32环境下，最主流、兼容性最好的依然是微软自家的Visual Studio。我使用的是Visual Studio 2022，社区版就完全够用。确保安装时勾选了“使用C++的桌面开发”工作负载，这会包含MSVC编译器、链接器和基本的Windows SDK。

除了VS，我们还需要几个辅助工具：

1. Git：用于克隆librdkafka的源代码。
2. CMake：这是编译librdkafka的关键。务必安装最新稳定版（如3.25+），并记得在安装时选择“将CMake添加到系统PATH”。
3. OpenSSL：librdkafka依赖OpenSSL进行加密和SASL认证。在Windows上获取OpenSSL开发库比较省事的方法是使用vcpkg（微软的C++库管理器），或者直接下载预编译的二进制包。为了流程清晰，我这里采用直接下载的方式。我们可以从 slproweb.com 下载适合的Win32 OpenSSL安装包（例如Win32 OpenSSL v1.1.1w Light）。安装后，记住它的安装路径，比如C:\Program Files (x86)\OpenSSL-Win32。

打开一个x86 Native Tools Command Prompt for VS 2022（注意是x86，对应Win32）。这个命令行工具非常重要，它配置好了所有VS的编译环境变量。后续的所有命令都在这个窗口下执行。

2.2 编译librdkafka静态库

我们不推荐直接使用动态库（DLL），在Win32 C++项目中，静态链接能减少部署依赖，避免运行时找不到DLL的尴尬。

# 1. 克隆代码（如果慢，可以找国内镜像） git clone https://github.com/confluentinc/librdkafka.git cd librdkafka # 2. 使用CMake配置并生成VS解决方案 mkdir build.win32 cd build.win32 cmake -G "Visual Studio 17 2022" -A Win32 ..

这里解释一下参数：-G指定生成器，-A Win32指定目标平台为32位。执行成功后，会在build.win32目录下生成librdkafka.sln解决方案文件。

接下来需要告诉CMake OpenSSL的位置。如果CMake没有自动找到，你需要手动指定。更稳妥的做法是在CMake命令中直接设置路径：

# 假设OpenSSL安装在默认路径 cmake -G "Visual Studio 17 2022" -A Win32 -DOPENSSL_ROOT_DIR="C:\Program Files (x86)\OpenSSL-Win32" ..

注意：路径中如果有空格，必须用双引号括起来。如果遇到找不到OpenSSL的错误，请仔细检查路径是否正确，以及安装的OpenSSL是否是Win32版本。

配置成功后，用VS编译：

# 3. 编译Release版本的静态库 cmake --build . --config Release --target rdkafka

--target rdkafka指定只编译核心的librdkafka库。编译完成后，你需要的核心产出物在build.win32\src\Release目录下：

• rdkafka.lib：静态库文件。
• rdkafka.h等头文件在源码的src目录下。

2.3 整理开发所需文件

为了在VS项目中方便引用，我习惯创建一个第三方库目录，比如D:\Dev\ThirdParty\librdkafka，然后把必要的文件整理过去：

librdkafka/ ├── include/ │ ├── rdkafka.h │ └── rdkafkacpp.h (如果你需要用C++接口) ├── lib/ │ └── Win32/ │ └── rdkafka.lib └── licenses/

把src目录下的rdkafka.h和rdkafkacpp.h拷贝到include。把编译好的rdkafka.lib拷贝到lib\Win32。这样结构清晰，后续项目配置时一目了然。

3. Visual Studio项目配置实战

库编译好了，接下来就是在你的C++项目中引入它。这里以创建一个新的Win32控制台项目为例。

3.1 创建项目与基础配置

打开VS2022，创建新项目 -> “控制台应用”（C++），项目名称比如KafkaWin32Demo。创建后，在解决方案资源管理器中，右键项目 -> “属性”。我们需要配置的是所有配置和Win32平台，避免Debug和Release切换时重复设置。

首先，配置头文件包含路径：

1. 在“C/C++” -> “常规” -> “附加包含目录”中，添加你的librdkafka头文件路径，例如D:\Dev\ThirdParty\librdkafka\include。

然后，配置库文件路径和链接库： 2. 在“链接器” -> “常规” -> “附加库目录”中，添加你的lib文件路径，例如D:\Dev\ThirdParty\librdkafka\lib\Win32。 3. 在“链接器” -> “输入” -> “附加依赖项”中，添加rdkafka.lib;ws2_32.lib;crypt32.lib。 -rdkafka.lib是我们刚编译的库。 -ws2_32.lib是Windows sockets库，网络通信必需。 -crypt32.lib是加密API库，OpenSSL依赖它。

3.2 解决潜在的运行时依赖

虽然我们链接的是静态库，但librdkafka和OpenSSL本身可能依赖一些动态库。最关键的是OpenSSL的运行时DLL。你需要将OpenSSL安装目录下的bin文件夹（例如C:\Program Files (x86)\OpenSSL-Win32\bin）中的libcrypto-1_1.dll和libssl-1_1.dll拷贝到你的项目生成可执行文件（.exe）的同一目录下，否则程序启动时会报“找不到指定模块”的错误。

一个更工程化的做法是，在项目属性 -> “生成事件” -> “后期生成事件”中，添加一个命令行，自动拷贝这些DLL到输出目录：

xcopy /Y “C:\Program Files (x86)\OpenSSL-Win32\bin\*.dll” “$(OutDir)”

这样每次编译后，DLL都会自动到位。

3.3 第一个连接测试：获取Kafka版本

在深入生产消费之前，我们先写个最简单的程序验证环境是否搭通。这能快速排除配置错误。

#include <iostream> #include <rdkafka.h> int main() { // 创建一个简单的配置对象 rd_kafka_conf_t* conf = rd_kafka_conf_new(); // 获取并打印librdkafka的版本 std::cout << "librdkafka version: " << rd_kafka_version_str() << std::endl; std::cout << "librdkafka version (hex): 0x" << std::hex << rd_kafka_version() << std::dec << std::endl; // 清理配置对象 rd_kafka_conf_destroy(conf); std::cout << "Environment test passed!" << std::endl; return 0; }

编译并运行这个程序。如果成功输出类似librdkafka version: 2.2.0的信息，那么恭喜你，最艰难的环境配置已经成功了。如果遇到链接错误或运行时崩溃，请回头仔细检查库路径、附加依赖项以及OpenSSL DLL是否到位。

4. Kafka生产者（Producer）核心实现

环境通了，我们来点实际的。生产者负责发送消息到Kafka。在Win32 C++环境下，我们需要关注几个核心点：配置的设定、消息的构造、发送的异步回调以及资源的妥善管理。

4.1 生产者配置与创建

生产者的行为由一系列配置参数控制。以下是一些最关键的配置，我习惯用一个辅助函数来创建基础配置：

#include <string> #include <rdkafka.h> rd_kafka_conf_t* create_producer_config(const std::string& brokers) { rd_kafka_conf_t* conf = rd_kafka_conf_new(); char errstr[512]; // 1. 设置Broker地址列表（必须） if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) { std::cerr << "Failed to set bootstrap.servers: " << errstr << std::endl; rd_kafka_conf_destroy(conf); return nullptr; } // 2. 设置消息发送确认机制（可靠性关键） // “all” 表示消息需要被所有ISR（同步副本）确认，是最强的一致性保证。 if (rd_kafka_conf_set(conf, "acks", "all", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) { std::cerr << "Failed to set acks: " << errstr << std::endl; // 错误处理... } // 3. 设置生产者ID，便于监控和调试 if (rd_kafka_conf_set(conf, "client.id", "win32_cpp_producer", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) { // 错误处理... } // 4. 设置消息发送失败后的重试次数和间隔 if (rd_kafka_conf_set(conf, "retries", "3", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {} if (rd_kafka_conf_set(conf, "retry.backoff.ms", "100", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {} // 5. 设置消息压缩方式，提升网络效率（可选，snappy较通用） if (rd_kafka_conf_set(conf, "compression.type", "snappy", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {} return conf; }

配置完成后，就可以创建生产者实例了：

rd_kafka_t* create_kafka_producer(rd_kafka_conf_t* conf) { char errstr[512]; rd_kafka_t* rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr)); if (!rk) { std::cerr << "Failed to create producer: " << errstr << std::endl; // 注意：如果创建失败，conf对象已被销毁或由函数接管，这里不应再destroy return nullptr; } // 添加Broker地址（也可以在配置中用bootstrap.servers，这里是一种替代方式） // rd_kafka_brokers_add(rk, brokers.c_str()); return rk; }

实操心得：rd_kafka_new调用后，配置对象conf的所有权就转移给了rk。之后你不能再使用或销毁conf，否则会导致未定义行为。这是一个容易踩坑的地方。

4.2 消息构造与异步发送

Kafka消息由键（Key）、值（Value）和可选头部（Headers）组成。在C接口中，我们使用rd_kafka_producev函数来发送，它支持可变参数，非常灵活。

bool produce_message(rd_kafka_t* rk, const std::string& topic, int partition, const std::string& key, const void* value, size_t val_len) { rd_kafka_resp_err_t err; // 使用rd_kafka_producev发送消息 err = rd_kafka_producev( rk, RD_KAFKA_V_TOPIC(topic.c_str()), // 主题 RD_KAFKA_V_PARTITION(partition), // 分区，RD_KAFKA_PARTITION_UA表示由分区器决定 RD_KAFKA_V_KEY(key.data(), key.size()), // 消息键 RD_KAFKA_V_VALUE(value, val_len), // 消息值 RD_KAFKA_V_END // 参数结束标志 ); if (err) { std::cerr << "Failed to produce message: " << rd_kafka_err2str(err) << std::endl; return false; } // 重要：触发轮询，确保发送回调被调用 rd_kafka_poll(rk, 0); return true; }

调用示例：

std::string topic = "test-topic"; std::string message_key = "device-001"; std::string message_value = "{\"timestamp\": 1698301200, \"status\": \"ok\"}"; if (!produce_message(producer, topic, RD_KAFKA_PARTITION_UA, message_key, message_value.data(), message_value.size())) { // 处理发送失败 }

这里RD_KAFKA_PARTITION_UA表示使用默认的分区器。如果键（Key）不为空，默认分区器会对键进行哈希，确保相同键的消息总是去到同一个分区，这对于保证相同键的消息顺序性至关重要。

4.3 发送回调（Delivery Report）与资源清理

异步发送后，我们怎么知道消息是否成功送达Kafka Broker？这就需要设置发送回调（Delivery Report Callback）。

首先，定义一个回调函数：

void dr_msg_cb(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage, void* opaque) { if (rkmessage->err) { // 发送失败 std::cerr << "Message delivery failed: " << rd_kafka_err2str(rkmessage->err) << std::endl; // 这里可以实现重试逻辑 } else { // 发送成功 std::cout << "Message delivered to " << rd_kafka_topic_name(rkmessage->rkt) << " [" << rkmessage->partition << "] at offset " << rkmessage->offset << std::endl; } // 注意：回调函数中不要释放rkmessage，librdkafka会处理。 }

然后，在创建配置对象之后，创建生产者实例之前，将这个回调设置到配置里：

rd_kafka_conf_set_dr_msg_cb(conf, dr_msg_cb);

发送回调是异步的，由rd_kafka_poll()函数驱动。因此，在主循环或发送消息后，需要定期调用rd_kafka_poll(rk, timeout_ms)来触发回调。timeout_ms设为0表示非阻塞，立即返回。

最后，程序退出时，必须妥善清理资源，确保所有在途消息的回调都被处理：

void cleanup_producer(rd_kafka_t* rk) { if (!rk) return; // 1. 刷新生产者，等待所有在途消息完成（发送或超时） // 参数是最大等待毫秒数 rd_kafka_flush(rk, 10 * 1000); // 等待10秒 // 2. 销毁生产者实例，这会自动销毁关联的配置和主题对象 rd_kafka_destroy(rk); std::cout << "Producer cleaned up." << std::endl; }

踩坑记录：直接调用rd_kafka_destroy而不调用flush，可能会导致还在内存队列或网络缓冲区的消息丢失，且它们的发送回调永远不会被调用。务必先刷新再销毁。

5. Kafka消费者（Consumer）核心实现

消费者从Kafka拉取消息。在Win32 C++中，我们需要处理订阅、拉取循环、偏移量提交和消费者组协调等问题。

5.1 消费者配置与创建

消费者的配置与生产者有重叠，也有其特有的设置。

rd_kafka_conf_t* create_consumer_config(const std::string& brokers, const std::string& group_id) { rd_kafka_conf_t* conf = rd_kafka_conf_new(); char errstr[512]; // 1. Broker地址和客户端ID rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), errstr, sizeof(errstr)); rd_kafka_conf_set(conf, "client.id", "win32_cpp_consumer", errstr, sizeof(errstr)); // 2. 消费者组ID（必须！用于偏移量管理和负载均衡） rd_kafka_conf_set(conf, "group.id", group_id.c_str(), errstr, sizeof(errstr)); // 3. 偏移量重置策略（当没有初始偏移量或偏移量失效时） // “earliest”: 从最早的消息开始消费 // “latest”: 从最新的消息开始消费（默认） rd_kafka_conf_set(conf, "auto.offset.reset", "earliest", errstr, sizeof(errstr)); // 4. 是否自动提交偏移量（建议先关闭，手动控制以保准确认） rd_kafka_conf_set(conf, "enable.auto.commit", "false", errstr, sizeof(errstr)); // 5. 自动提交间隔（如果enable.auto.commit=true） // rd_kafka_conf_set(conf, "auto.commit.interval.ms", "5000", errstr, sizeof(errstr)); // 6. 每次poll最大拉取的消息字节数 rd_kafka_conf_set(conf, "fetch.max.bytes", "1048576", errstr, sizeof(errstr)); // 1MB // 7. 最大拉取间隔，超时则broker认为消费者已死 rd_kafka_conf_set(conf, "session.timeout.ms", "10000", errstr, sizeof(errstr)); return conf; }

创建消费者实例使用RD_KAFKA_CONSUMER类型：

rd_kafka_t* create_kafka_consumer(rd_kafka_conf_t* conf) { char errstr[512]; rd_kafka_t* rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr)); if (!rk) { std::cerr << "Failed to create consumer: " << errstr << std::endl; return nullptr; } return rk; }

5.2 订阅主题与消息拉取循环

创建消费者后，需要订阅一个或多个主题。然后进入一个主循环，不断拉取（poll）消息。

bool subscribe_to_topic(rd_kafka_t* rk, const std::vector<std::string>& topics) { rd_kafka_topic_partition_list_t* subscription = rd_kafka_topic_partition_list_new(topics.size()); for (const auto& topic : topics) { rd_kafka_topic_partition_list_add(subscription, topic.c_str(), RD_KAFKA_PARTITION_UA); } rd_kafka_resp_err_t err = rd_kafka_subscribe(rk, subscription); rd_kafka_topic_partition_list_destroy(subscription); if (err) { std::cerr << "Failed to subscribe: " << rd_kafka_err2str(err) << std::endl; return false; } std::cout << "Subscribed to topics successfully." << std::endl; return true; }

订阅成功后，就可以开始消费循环了：

void consumer_loop(rd_kafka_t* rk, int timeout_ms) { bool running = true; while (running) { // rd_kafka_consumer_poll 是核心消费函数 rd_kafka_message_t* rkmessage = rd_kafka_consumer_poll(rk, timeout_ms); if (!rkmessage) { // 超时，没有消息，继续循环 continue; } if (rkmessage->err) { // 这是一个错误（例如分区结束、偏移量无效等） if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { // 已到达分区末尾，暂时没有新消息 std::cout << "Reached end of partition " << rkmessage->partition << std::endl; } else { // 其他错误 std::cerr << "Consumer error: " << rd_kafka_message_errstr(rkmessage) << std::endl; // 根据错误类型决定是否退出循环 if (rkmessage->err == RD_KAFKA_RESP_ERR__TRANSPORT) { // 网络错误，可能需要重建消费者 running = false; } } // 错误消息也需要释放 rd_kafka_message_destroy(rkmessage); continue; } // 成功收到消息 process_kafka_message(rkmessage); // 处理完消息后，手动提交偏移量（异步） // 注意：提交的是当前消息的偏移量+1，表示已处理到此位置 rd_kafka_resp_err_t commit_err; commit_err = rd_kafka_commit_message(rk, rkmessage, 0); // 0表示异步提交 if (commit_err) { std::cerr << "Failed to commit offset: " << rd_kafka_err2str(commit_err) << std::endl; } // 释放消息资源 rd_kafka_message_destroy(rkmessage); } } void process_kafka_message(const rd_kafka_message_t* rkmessage) { std::string topic_name = rd_kafka_topic_name(rkmessage->rkt); int partition = rkmessage->partition; int64_t offset = rkmessage->offset; // 处理消息键 std::string key_str; if (rkmessage->key) { key_str.assign(static_cast<const char*>(rkmessage->key), rkmessage->key_len); } // 处理消息值（业务负载） std::string value_str; if (rkmessage->payload) { value_str.assign(static_cast<const char*>(rkmessage->payload), rkmessage->len); } std::cout << "Consumed message: " << "Topic[" << topic_name << "], " << "Partition[" << partition << "], " << "Offset[" << offset << "], " << "Key[" << key_str << "], " << "Value: " << value_str.substr(0, 100) << "..." << std::endl; // 只打印前100字符 // 这里添加你的实际业务处理逻辑 // ... }

关键点解析：rd_kafka_consumer_poll是阻塞调用，参数timeout_ms指定了最长等待时间。如果设为1000，那么最多等待1秒，即使没有消息也会返回NULL。这给了你在消费循环中插入其他逻辑（如检查退出标志）的机会。另外，偏移量提交是保证“至少一次”或“恰好一次”语义的关键。异步提交性能好，但可能在消费者崩溃时丢失少量消息。对于严格场景，可以使用同步提交（rd_kafka_commit_message(rk, rkmessage, 1)），但会降低吞吐。

5.3 消费者关闭与偏移量提交

优雅关闭消费者同样重要，需要确保退出前提交最后的偏移量，避免重复消费。

void cleanup_consumer(rd_kafka_t* rk) { if (!rk) return; // 1. 关闭消费者，停止拉取消息并离开消费者组 // 这会触发一次最终的偏移量提交（如果enable.auto.commit=true） rd_kafka_consumer_close(rk); // 2. 如果手动提交，为了保险，可以再显式刷新一下 // rd_kafka_commit(rk, NULL, 0); // 同步提交所有分配的分区 // 3. 销毁消费者实例 rd_kafka_destroy(rk); std::cout << "Consumer cleaned up." << std::endl; }

6. 高级配置、性能调优与问题排查

基础的生产消费跑通后，我们还需要关注一些高级特性和性能问题，让程序更健壮、更高效。

6.1 关键配置参数深度解析

librdkafka的配置参数多达上百个，这里挑几个在Win32环境下需要特别关注的：

6.2 Win32环境下的性能与稳定性调优

1. 内存管理：长时间运行的生产者/消费者，要注意内存泄漏。确保每个rd_kafka_message_t*在使用后都调用rd_kafka_message_destroy。定期检查rd_kafka_mem_*系列函数（如果编译时开启了统计）来监控内存使用。
2. 线程安全：librdkafka的API大部分是线程安全的，但像rd_kafka_t对象本身，其生命周期管理（创建、销毁）最好在单一主线程进行。生产/消费的调用可以从不同线程进行。在Win32多线程程序中，注意使用适当的同步原语。
3. 网络与超时：Win32的网络环境可能比Linux更复杂（如企业防火墙、代理）。如果遇到连接问题，可以调大socket.timeout.ms（默认30秒）和connections.max.idle.ms。使用debug配置项（如debug=broker,protocol）可以打印详细的网络通信日志，但会严重影响性能，仅用于调试。
4. CPU占用：消费者的poll循环如果timeout_ms设置过小（如0），会导致空转，CPU占用率飙升。通常设置为100-1000ms是一个合理的范围，在响应速度和CPU占用间取得平衡。

6.3 常见问题与排查技巧实录

在实际开发中，我遇到了不少问题，这里总结几个典型的：

问题1：生产者发送消息成功，但消费者收不到。

• 排查步骤：
  1. 检查消费者组ID和偏移量：确认消费者是否使用了新的组ID，导致从最新偏移量（latest）开始消费，而错过了历史消息。可以尝试将auto.offset.reset改为earliest，或者换一个全新的组ID。
  2. 检查Topic和分区：确认生产者和消费者订阅的是同一个Topic。用Kafka命令行工具（如kafka-console-consumer.bat）直接消费，看是否有数据。
  3. 检查Broker地址：确保生产者和消费者配置的bootstrap.servers是正确的，并且网络可达。
  4. 开启调试日志：在生产者配置中设置debug=msg，在消费者配置中设置debug=cgrp,topic,fetch，观察输出。

问题2：程序崩溃在rd_kafka_new或rd_kafka_producev。

• 排查步骤：
  1. 库不匹配：确保编译librdkafka的运行时环境（如VC++ Redistributable版本）与你的应用程序匹配。Debug/Release模式也要一致。
  2. 内存损坏：检查是否有数组越界、野指针等问题，这些可能在调用librdkafka前就破坏了堆栈。
  3. 配置对象生命周期：确认传递给rd_kafka_new的配置对象conf没有被提前销毁或重复使用。

问题3：消费者拉取消息延迟很高，或者吞吐量上不去。

• 排查步骤：
  1. 调整fetch.max.bytes和max.partition.fetch.bytes：如果消息体较大，默认的1MB可能不够，导致一次poll只拉回少量消息。
  2. 检查fetch.wait.max.ms：如果设置过大，在低流量Topic上会人为增加延迟。
  3. 并行度不足：单个消费者线程消费多个分区可能成为瓶颈。可以考虑为每个分区启动一个独立的消费者线程（但属于同一个消费者组），或者使用rd_kafka_consumer_poll的并行调用（需要仔细管理分区分配）。
  4. 业务处理瓶颈：检查process_kafka_message函数是否耗时过长。如果业务处理慢，消息会堆积在客户端。考虑将业务处理放入独立线程池。

问题4：如何优雅地处理程序退出（如Ctrl+C）？在Win32控制台程序中，可以设置控制台控制处理器（Console Control Handler）来捕获中断信号。

#include <Windows.h> static volatile sig_atomic_t run = 1; BOOL WINAPI ConsoleHandler(DWORD signal) { if (signal == CTRL_C_EVENT) { run = 0; return TRUE; } return FALSE; } int main() { SetConsoleCtrlHandler(ConsoleHandler, TRUE); // ... 初始化生产者/消费者 ... while (run) { // 生产或消费循环 // 在循环内定期检查 run 变量 } // ... 调用 cleanup_producer/cleanup_consumer ... return 0; }

这样，当用户按下Ctrl+C时，run标志会被置零，主循环退出，然后执行清理逻辑，确保偏移量提交和资源释放。

最后，再分享一个调试小技巧：将librdkafka的日志输出到文件，便于离线分析。在配置中设置log_level和log.queue，并实现一个日志回调函数（rd_kafka_conf_set_log_cb），将日志写入文件或标准错误。这对于排查线上问题非常有帮助。整个流程走下来，虽然Win32下配置稍显复杂，但一旦打通，librdkafka提供的稳定性和高性能绝对值得投入。