c++ 多线程kafla实现 消费者消费时间延迟一分钟

可以使用C++11标准中的`<thread>`头文件来实现多线程，同时使用Kafka提供的Consumer API来消费消息。

在消费者中，可以使用`std::this_thread::sleep_for()`函数来实现时间延迟一分钟。具体实现代码如下：

#include <iostream>
#include <thread>
#include <chrono>
#include <librdkafka/rdkafkacpp.h>

class KafkaConsumer {
public:
    KafkaConsumer(const std::string& brokers, const std::string& topic)
        : brokers_(brokers), topic_(topic) {
        // 创建配置对象
        RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

        // 设置Broker列表
        std::string errstr;
        if (conf->set("metadata.broker.list", brokers_, errstr) != RdKafka::Conf::CONF_OK) {
            std::cerr << "Failed to set broker list: " << errstr << std::endl;
            exit(1);
        }

        // 创建KafkaConsumer对象
        consumer_ = RdKafka::Consumer::create(conf, errstr);
        if (!consumer_) {
            std::cerr << "Failed to create Kafka consumer: " << errstr << std::endl;
            exit(1);
        }

        // 订阅Topic
        RdKafka::ErrorCode err = consumer_->subscribe({ topic_ });
        if (err != RdKafka::ERR_NO_ERROR) {
            std::cerr << "Failed to subscribe to topic " << topic_ << ": "
                      << RdKafka::err2str(err) << std::endl;
            exit(1);
        }
    }

    void consume() {
        while (true) {
            RdKafka::Message* msg = consumer_->consume(1000);
            if (msg->err() == RdKafka::ERR_NO_ERROR) {
                std::cout << "Received message: " << std::string(static_cast<char*>(msg->payload()), msg->len()) << std::endl;
            }
            delete msg;

            // 延迟一分钟
            std::this_thread::sleep_for(std::chrono::minutes(1));
        }
    }

private:
    std::string brokers_;
    std::string topic_;
    RdKafka::Consumer* consumer_;
};

int main() {
    KafkaConsumer consumer("localhost:9092", "test_topic");
    consumer.consume();
    return 0;
}

在上面的代码中，`consume()`函数是消费消息的主要逻辑，其中调用了`std::this_thread::sleep_for(std::chrono::minutes(1))`来实现时间延迟一分钟的操作。同时，为了简化代码，省略了一些异常处理和资源释放的代码。