我想写一个客户作为制片人。我按照示例创建了一个新的win32控制台项目。我发现api不适合我，除非我在程序末尾添加getline（）函数。
如果删除getline（），product（..）方法仍然返回success的结果。但是，我在kafka控制台的命令窗口中看不到任何响应
我有点糊涂了。是这样吗？不使用getline（）如何发送消息？有人知道吗？
我找到了为什么它不起作用。删除producer对象似乎太快，导致producer无法向代理发送消息。
当我在produce方法和delete producer对象之间添加sleep 1000时，producer可以正确地发送消息。
所以，问题是如何立即发送消息。在销毁producer对象之前，如何确保这些消息已完全发送？
如何解决这个问题，实际上我不喜欢在源代码中添加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;
}