【问题标题】:Kafka C++ client taking a long time to receive a messageKafka C ++客户端需要很长时间才能收到消息
【发布时间】:2020-06-08 14:51:45
【问题描述】:

我正在使用 cppkafka library,它是 librdkafka 的包装器,而 C++ Kafka 客户端则用于非常简单的消息流任务。我的消费者类行为怪异,因为接收消息需要相当长的时间。更准确地说,每次接收可执行文件运行并保持运行时,消费者可以正确接收第一批消息,但后续消息大约需要 15 秒才能到达。任何人都知道什么可能性会导致这样的事情(kafka 配置、库特定问题或我的愚蠢错误)?一百万谢谢。

我的接收线程如下

configuration_.set("group.id", 0);
consumer_ = std::make_unique<cppkafka::Consumer>(configuration_);
consumer_->subscribe({TopicTraits<trade::OrderRequest>::topic, TopicTraits<trade::CancelRequest>::topic});
std::thread([this] {
  while (working_) {
    cppkafka::Message msg = consumer_->poll();
    if (msg) {
      if (msg.get_error()) {
        if (!msg.is_eof()) {
          ERROR("error occurred while polling message: {}", msg.get_error());
        }
      } else {
        try {
          Json j = Json::parse(msg.get_payload());
          if (msg.get_topic() == TopicTraits<trade::OrderRequest>::topic) {
            INFO("received [order_req], {}", msg.get_payload());
            ReceiveOrderRequest(j.get<trade::OrderRequest>());
          } else if (msg.get_topic() == TopicTraits<trade::CancelRequest>::topic) {
            INFO("received [cancel_req], {}", msg.get_payload());
            ReceiveCancelRequest(j.get<trade::CancelRequest>());
          }
        } catch (const std::exception &e) {
          ERROR("error occurred while handling incoming message, {}", e.what());
        }
      }
    }
  }
}).detach();

【问题讨论】:

  • 由于我还在开发系统,不可能是消息在队列中堆积。我只用了几条相当短的消息进行测试。

标签: c++ apache-kafka librdkafka


【解决方案1】:

订阅不同主题的相同组 id 的两个消费者阻止 poll()

经过一番研究,我发现问题与 kafka 的一个更基本的配置选项有关。问题是我的消费者在调用 poll() 时被阻塞,直接原因是两个具有相同组 id 的消费者订阅了不同的主题。我重新分配了组 ID,问题就消失了。

【讨论】:

    猜你喜欢
    • 2018-03-18
    • 1970-01-01
    • 2019-05-27
    • 1970-01-01
    • 1970-01-01
    • 2017-06-17
    • 1970-01-01
    • 1970-01-01
    • 2018-01-20
    相关资源
    最近更新 更多