c++ kafka_kafka c++

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

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

```bash
sudo apt-get install librdkafka-dev
```

### 生产者示例

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

```cpp
#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` 并消费其中的消息：

```cpp
#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 进行消息的生产和消费。