c++ kafka

在 C++ 中使用 Kafka，通常我们会使用 Confluent 的 Kafka C++ 客户端库。下面我将给出一个简单的示例，展示如何使用这个库来创建一个生产者和一个消费者。

首先，你需要安装 Confluent 的 Kafka C++ 客户端。你可以从 GitHub 上获取源代码并编译，或者如果你使用的是 Debian 或 Ubuntu，可以使用下面的命令安装：

 sudo apt-get install librdkafka-dev

生产者示例

这是一个简单的 Kafka 生产者的示例代码，它会向一个名为 test_topic 的主题发送消息：

 #include <iostream>
 #include <rdkafka/rdkafkacpp.h>
void delivery_report(RdKafka::Message &message) {
   if (message.err() == RdKafka::ERR__TIMED_OUT)
     std::cout << "Message delivery timed out." << std::endl;
   else if (message.err())
     std::cerr << "Failed to deliver message: " << message.errstr() << std::endl;
   else
     std::cout << "Produced message to topic " << message.topic_name()
               << " at offset " << message.offset() << std::endl;
 }
int main() {
   std::string errstr;
  // 创建一个生产者配置实例
   RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  // 设置 broker 列表
   conf->set("bootstrap.servers", "localhost:9092", errstr);
   if (conf->get_err()) {
     std::cerr << "Failed to set 'bootstrap.servers' configuration property: "
               << conf->errstr() << std::endl;
     return -1;
   }
  // 创建一个生产者实例
   RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
   if (!producer) {
     std::cerr << "Failed to create producer: " << errstr << std::endl;
     return -1;
   }
  // 发送消息
   const char *topic = "test_topic";
   const char *key = "mykey";
   const char *payload = "Hello, Kafka!";
   RdKafka::ErrorCode resp = producer->produce(
       topic, RdKafka::TopicPartition::PARTITION_UA, RdKafka::Producer::RK_MSG_COPY,
       const_cast<char *>(payload), strlen(payload), const_cast<char *>(key),
       strlen(key), NULL, delivery_report);
   if (resp != RdKafka::ERR_NO_ERROR) {
     std::cerr << "% Produce failed: " << RdKafka::err2str(resp) << std::endl;
   }
  // 等待所有消息发送完成
   producer->poll(0);
  // 清理
   delete producer;
   delete conf;
  return 0;
 }

消费者示例

接下来是一个简单的 Kafka 消费者的示例代码，它订阅了 test_topic 并消费其中的消息：

 #include <iostream>
 #include <rdkafka/rdkafkacpp.h>
class ConsumerRebalanceCb : public RdKafka::RebalanceCb {
 public:
   void on_rebalance(RdKafka::KafkaConsumer *consumer,
                     RdKafka::ErrorCode err, std::vector<RdKafka::TopicPartition *> &partitions, void *opaque) {
     std::cerr << "Rebalance triggered: " << RdKafka::err2str(err) << std::endl;
     if (err == RdKafka::ERR__ASSIGN_PARTITIONS)
       consumer->assign(partitions);
     else
       consumer->unassign();
   }
 };
int main() {
   std::string errstr;
   RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
   conf->set("bootstrap.servers", "localhost:9092", errstr);
   if (conf->get_err()) {
     std::cerr << "Failed to set 'bootstrap.servers' configuration property: "
               << conf->errstr() << std::endl;
     return -1;
   }
   conf->set("group.id", "test_group", errstr);
   if (conf->get_err()) {
     std::cerr << "Failed to set 'group.id' configuration property: "
               << conf->errstr() << std::endl;
     return -1;
   }
  RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);
   if (!consumer) {
     std::cerr << "Failed to create consumer: " << errstr << std::endl;
     return -1;
   }
  ConsumerRebalanceCb rebalance_cb;
   consumer->set_rebalance_cb(rebalance_cb);
  std::vector<std::string> topics;
   topics.push_back("test_topic");
   consumer->subscribe(topics);
  bool running = true;
   while (running) {
     RdKafka::Message *msg = consumer->consume(1000);
     switch (msg->err()) {
       case RdKafka::ERR_NO_ERROR:
         std::cout << "%% Message on " << msg->topic_name()
                   << " at offset " << msg->offset() << ": "
                   << (char *)msg->payload() << std::endl;
         break;
       case RdKafka::ERR__PARTITION_EOF:
         std::cerr << "%% Reached end of partition" << std::endl;
         break;
       default:
         std::cerr << "%% Error " << msg->err() << ": "
                   << msg->errstr() << std::endl;
         break;
     }
     delete msg;
   }
  consumer->unsubscribe();
  delete consumer;
   delete conf;
  return 0;
 }

请确保将上述代码中的 localhost:9092 替换为你实际的 Kafka 服务器地址。此外，你也需要根据你的需求调整 topic 名称、组 ID 等参数。这两个示例展示了如何在 C++ 中使用 Kafka 进行消息的生产和消费。