如何在librdkafka中使用KAFKA的生成器API和Windows上的c ++代码

Question

我正在尝试将客户端编写为生产者.我按照示例创建了一个新的win32控制台项目.我发现API不适用于我,除非我在程序结束时添加了getline()函数.

如果我删除了getline(),那么produce(..)方法仍然会返回成功的结果.但是,我在kafka-console-consumer的命令窗口中看不到任何响应

我有点困惑.是对的吗？如何在不使用getline()的情况下发送消息？有谁知道？

我发现为什么它不起作用.删除生产者对象似乎太快,导致生产者无法向代理发送消息.

当我在Produ方法和删除生成器对象之间添加sleep 1000时,生产者可以正确地发送消息.

所以,问题是如何立即发送消息.在销毁生产者对象之前,如何确保完全发送这些消息？

如何解决这个问题,实际上我不喜欢在我的源代码中添加一些sleep().

win10 + vs2015 + kafka_2.10-0.9.0.1 + zookeeper-3.4.6 + librdkafka请查看以下代码

    // kafka_test_win32_nomfc.cpp 
//

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


int static producer_1()
{
    std::string brokers = "127.0.0.1";
    std::string errstr;
    std::string topic_str = "linli";
    std::string mode;
    std::string debug;
    int32_t partition = RdKafka::Topic::PARTITION_UA;
    int64_t start_offset = RdKafka::Topic::OFFSET_BEGINNING;
    bool do_conf_dump = false;
    int opt;
    // MyHashPartitionerCb hash_partitioner;
    int use_ccb = 0;

    /*
    * Create configuration objects
    */
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

    conf->set("metadata.broker.list", brokers, errstr);

    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        exit(1);
    }

    std::cout << "% Created producer " << producer->name() << std::endl;

    /*
    * Create topic handle.
    */
    RdKafka::Topic *topic = NULL;
    if (!topic_str.empty()) {
        topic = RdKafka::Topic::create(producer, topic_str, tconf, errstr);
        if (!topic) {
            std::cerr << "Failed to create topic: " << errstr << std::endl;
            exit(1);
        }
    }

    RdKafka::ErrorCode resp = producer->produce(topic, partition,
        RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
        const_cast<char *>("hello worlf"), 11,
        NULL, NULL);

    delete topic;
    delete producer;
    return 0;
}


int static producer_2()
{
    std::string brokers = "127.0.0.1";
    std::string errstr;
    std::string topic_str = "linli";
    std::string mode;
    std::string debug;
    int32_t partition = RdKafka::Topic::PARTITION_UA;
    int64_t start_offset = RdKafka::Topic::OFFSET_BEGINNING;
    bool do_conf_dump = false;
    int opt;
    // MyHashPartitionerCb hash_partitioner;
    int use_ccb = 0;

    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

    conf->set("metadata.broker.list", brokers, errstr);

    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        exit(1);
    }

    std::cout << "% Created producer " << producer->name() << std::endl;

    RdKafka::ErrorCode resp = producer->produce(topic_str, partition,
        RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
        (void *)"hi", 2,
        NULL, 0, 0, NULL);



    std::string errs(RdKafka::err2str(resp));
    std::cout << errs << std::endl;
    //producer->poll(0);


    delete producer;

    return 0;
}


int main()
{

    producer_2();

    return 0;
}

Answer 1

librdkafka produce()API(包括C和C++)是异步的,您的消息最初只会在内部生成器队列中排队,并且稍后(请参阅queue.buffering.max.ms配置属性 - 默认1秒)与其他消息组合成消息批处理( MessageSet)并从后台线程发送到代理.

您的程序调用produce()然后快速退出,早在后台生产者线程有机会将消息发送给代理之前,更不用说从代理接收确认.

要确保已发送所有未完成的消息,请flush()在终止应用程序之前致电.

如果您的申请是长期存在的,您应该poll()定期拨打电话,以便为您注册的任何传递报告回调提供服务.