tconf->set("auto.offset.reset", "earliest", errstr);什么作用

这行代码是使用 librdkafka 库中的 `Conf` 类设置 Kafka 消费者的参数。具体来说，`auto.offset.reset` 是 Kafka 消费者的一个配置参数，用于确定消费者在启动时从哪里开始读取消息。其可选值包括：

- `earliest`：从最早的可用消息开始消费。
- `latest`：从最新的消息开始消费。
- `none`：如果没有找到消费者组的初始偏移量，则抛出异常。
- 一个整数值：从指定的偏移量开始消费。

在这行代码中，我们将 `auto.offset.reset` 的值设置为 `earliest`，表示消费者在第一次启动时会从最早可用的消息开始消费。这通常适用于消费者组刚创建时或消费者之前没有消费消息的情况。

相关问题

rdkafka_example

rdkafka 是一款 C/C++ 实现的高性能消息队列系统，支持多种协议和编程语言。以下是一个使用 rdkafka 的简单示例：

#include <iostream>
#include <string>
#include <csignal>
#include <librdkafka/rdkafkacpp.h>

static bool run = true;

void sigterm_handler(int sig) {
    run = false;
}

class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message &message) {
        if (message.err()) {
            std::cerr << "Message delivery failed: " << message.errstr() << std::endl;
        } else {
            std::cout << "Message delivered to " << message.topic_name() << " [" << message.partition() << "] at offset " << message.offset() << std::endl;
        }
    }
};

int main() {
    std::string brokers = "localhost:9092";
    std::string topic = "test";
    std::string errstr;
    int32_t partition = RdKafka::Topic::PARTITION_UA;
    int64_t start_offset = RdKafka::Topic::OFFSET_BEGINNING;

    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("metadata.broker.list", brokers, errstr);
    RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
    tconf->set("auto.offset.reset", "earliest", errstr);

    RdKafka::Consumer *consumer = RdKafka::Consumer::create(conf, errstr);
    RdKafka::Topic *top = RdKafka::Topic::create(consumer, topic, tconf, errstr);

    ExampleDeliveryReportCb ex_dr_cb;
    conf->set("dr_cb", &ex_dr_cb, errstr);

    consumer->start(top, partition, start_offset);

    signal(SIGINT, sigterm_handler);
    signal(SIGTERM, sigterm_handler);

    while (run) {
        RdKafka::Message *msg = consumer->consume(top, partition, 1000);
        if (msg) {
            if (msg->err() == RdKafka::ERR_NO_ERROR) {
                std::cout << "Received message: " << std::string(static_cast<char*>(msg->payload()), msg->len()) << std::endl;
                consumer->commitAsync(msg);
            } else {
                std::cerr << "Error while consuming message: " << msg->errstr() << std::endl;
            }
            delete msg;
        }
    }

    consumer->stop(top, partition);
    consumer->close();
    delete top;
    delete consumer;
    delete tconf;
    delete conf;

    return 0;
}

这个示例是一个 Kafka 消费者，它从指定的 Kafka 集群中的指定主题消费消息，并将消费到的消息输出到控制台。在程序运行期间，如果接收到 SIGINT 或 SIGTERM 信号，程序会停止消费并退出。

kafka指定消费1分钟前的数据实现延迟消费 c++ 版本 完整程序

以下是一个使用 C++ 实现 Kafka 延迟消费的示例程序：

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

class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message &message) {
        if (message.err()) {
            std::cerr << "Message delivery failed: " << message.errstr() << std::endl;
        } else {
            std::cout << "Message delivered to topic " << message.topic_name() << " [" << message.partition() << "] at offset " << message.offset() << std::endl;
        }
    }
};

int main() {
    std::string brokers = "localhost:9092";
    std::string topic = "test_topic";
    std::string errstr;

    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("metadata.broker.list", brokers, errstr);

    RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
    tconf->set("auto.offset.reset", "earliest", errstr);

    ExampleDeliveryReportCb ex_dr_cb;
    conf->set("dr_cb", &ex_dr_cb, errstr);

    RdKafka::Consumer *consumer = RdKafka::Consumer::create(conf, errstr);
    RdKafka::Topic *rd_topic = RdKafka::Topic::create(consumer, topic, tconf, errstr);

    int64_t timestamp = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count() - 60000;

    RdKafka::ErrorCode resp = consumer->start(rd_topic, 0, timestamp);
    if (resp != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Failed to start consumer: " << RdKafka::err2str(resp) << std::endl;
        return 1;
    }

    while (true) {
        RdKafka::Message *msg = consumer->consume(rd_topic, 0, 1000);
        if (msg->err() == RdKafka::ERR_NO_ERROR) {
            std::cout << "Received message: " << std::string((char *) msg->payload()) << std::endl;
        } else if (msg->err() == RdKafka::ERR__PARTITION_EOF) {
            std::cerr << "Reached end of partition" << std::endl;
        } else {
            std::cerr << "Failed to consume message: " << msg->errstr() << std::endl;
        }
        delete msg;
    }

    return 0;
}

该程序使用 librdkafka 库连接到 Kafka，并创建一个消费者实例。然后，它使用 `consumer->start()` 函数指定从 1 分钟前开始消费数据。最后，它使用 `consumer->consume()` 函数接收消息并进行处理。

请注意，需要在编译时链接 librdkafka 库。